From 1a94cec822f3aec23f32397b9bbb9ccf88da37aa Mon Sep 17 00:00:00 2001 From: hupenglong1 Date: Wed, 20 May 2026 15:04:19 +0800 Subject: [PATCH 1/7] =?UTF-8?q?=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/api/server.py | 1470 ++++++++++++++- .../app/api/claw/training/pipeline/route.ts | 26 + .../claw/training/pipeline/stream/route.ts | 38 + .../api/claw/training/step-detail/route.ts | 34 + frontend/app/app/assistant.tsx | 205 ++- frontend/app/app/globals.css | 80 + .../assistant-ui/activity-panel.tsx | 125 +- .../app/components/assistant-ui/thread.tsx | 19 +- .../assistant-ui/threadlist-sidebar.tsx | 36 +- .../components/training/step-detail-sheet.tsx | 247 +++ .../training/training-mode-toggle.tsx | 60 + .../training/training-pipeline-panel.tsx | 480 +++++ frontend/app/lib/use-training-mode.ts | 228 +++ skills/model-iteration/README.md | 66 - skills/model-iteration/SKILL.md | 1587 +++-------------- .../model-iteration/{ => assets}/config.yaml | 2 +- .../knowledge/multi_command_rules.md | 0 .../knowledge/multiturn_continuity_rules.md | 57 + .../knowledge/navigation_routing_rules.md | 0 .../knowledge/travel_routing_rules.md | 4 +- skills/model-iteration/references/program.md | 1438 +++++++++++++++ .../scripts/prepare_and_train_sft.py | 15 +- .../scripts/sft_train_job.yaml.tpl | 117 -- .../scripts/submit_sft_via_cml.sh | 104 -- src/agent_types.py | 2 +- src/jupyter_runtime.py | 40 + src/session_store.py | 44 +- 27 files changed, 4831 insertions(+), 1693 deletions(-) create mode 100644 frontend/app/app/api/claw/training/pipeline/route.ts create mode 100644 frontend/app/app/api/claw/training/pipeline/stream/route.ts create mode 100644 frontend/app/app/api/claw/training/step-detail/route.ts create mode 100644 frontend/app/components/training/step-detail-sheet.tsx create mode 100644 frontend/app/components/training/training-mode-toggle.tsx create mode 100644 frontend/app/components/training/training-pipeline-panel.tsx create mode 100644 frontend/app/lib/use-training-mode.ts delete mode 100644 skills/model-iteration/README.md rename skills/model-iteration/{ => assets}/config.yaml (94%) rename skills/model-iteration/{ => references}/knowledge/multi_command_rules.md (100%) create mode 100644 skills/model-iteration/references/knowledge/multiturn_continuity_rules.md rename skills/model-iteration/{ => references}/knowledge/navigation_routing_rules.md (100%) rename skills/model-iteration/{ => references}/knowledge/travel_routing_rules.md (75%) create mode 100644 skills/model-iteration/references/program.md delete mode 100644 skills/model-iteration/scripts/sft_train_job.yaml.tpl delete mode 100644 skills/model-iteration/scripts/submit_sft_via_cml.sh diff --git a/backend/api/server.py b/backend/api/server.py index 206e082..64c4a95 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -7,8 +7,10 @@ chat, slash commands, and saved sessions, and serves the static SPA. from __future__ import annotations import asyncio +from contextlib import asynccontextmanager import csv import hashlib +import shlex import json import os import queue @@ -27,7 +29,7 @@ from urllib import error, request from urllib.parse import quote, unquote, urlparse from uuid import uuid4 -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, Request from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field @@ -998,8 +1000,9 @@ class SkillSyncRequest(BaseModel): account_id: str | None = None -class SessionTitleUpdate(BaseModel): - title: str = Field(min_length=1, max_length=80) +class SessionUpdate(BaseModel): + title: str | None = Field(default=None, max_length=80) + is_training: bool | None = None class FeishuAccountRequest(BaseModel): @@ -1037,7 +1040,16 @@ def _append_runtime_context(current: str | None, addition: str) -> str: # --------------------------------------------------------------------------- def create_app(state: AgentState) -> FastAPI: - app = FastAPI(title='Claw Code GUI', version='1.0') + @asynccontextmanager + async def lifespan(app: FastAPI): # noqa: ARG001 + scanner_task = asyncio.create_task(_watcher_scanner_loop(state)) + try: + yield + finally: + scanner_task.cancel() + _watcher_manager.cancel_all() + + app = FastAPI(title='Claw Code GUI', version='1.0', lifespan=lifespan) # ------------- static + index ------------------------------------------ app.mount( @@ -1528,6 +1540,7 @@ def create_app(state: AgentState) -> FastAPI: 'modified_at': _session_mtime(path), 'model': model_config.get('model'), 'usage': usage, + 'is_training': bool(data.get('is_training', False)), } ) return results @@ -1555,17 +1568,168 @@ def create_app(state: AgentState) -> FastAPI: @app.patch('/api/sessions/{session_id}') async def update_session( session_id: str, - payload: SessionTitleUpdate, + payload: SessionUpdate, account_id: str | None = None, ) -> dict[str, Any]: directory = state.account_paths(account_id)['sessions'] safe_id = _safe_session_id(session_id) - title = _clean_manual_session_title(payload.title) - if safe_id is None or title is None: - raise HTTPException(status_code=400, detail='Invalid session title') - if not _update_session_title(directory, safe_id, title): - raise HTTPException(status_code=404, detail='Session not found') - return {'session_id': safe_id, 'title': title} + if safe_id is None: + raise HTTPException(status_code=400, detail='Invalid session id') + if payload.title is None and payload.is_training is None: + raise HTTPException(status_code=400, detail='Nothing to update') + + result: dict[str, Any] = {'session_id': safe_id} + + if payload.title is not None: + title = _clean_manual_session_title(payload.title) + if title is None: + raise HTTPException(status_code=400, detail='Invalid session title') + if not _update_session_title(directory, safe_id, title): + raise HTTPException(status_code=404, detail='Session not found') + result['title'] = title + + if payload.is_training is not None: + value = bool(payload.is_training) + if not _update_session_training(directory, safe_id, value): + raise HTTPException(status_code=404, detail='Session not found') + result['is_training'] = value + + return result + + @app.get('/api/training/step-detail') + async def get_training_step_detail( + session_id: str, + step: str, + account_id: str | None = None, + ) -> Response: + sessions_dir = state.account_paths(account_id)['sessions'] + state_path = _session_state_path(sessions_dir, session_id) + state_entries, _ = _read_program_state(state_path) + run_dic = _resolve_run_dic_from_state(state_entries) + runtime = _jupyter_runtime_for_session(state, account_id, session_id) + artifact_paths = _step_artifact_paths(step, run_dic) + + if not artifact_paths: + payload = { + 'step': step, + 'run_dic': run_dic, + 'format': 'empty', + 'content': '', + 'source_path': None, + 'fetched_at_ms': int(time.time() * 1000), + 'error': '该 step 没有定义产物路径', + } + return Response( + content=json.dumps(payload, ensure_ascii=False), + media_type='application/json', + headers={'Cache-Control': 'no-store'}, + ) + + last_error: str | None = None + for path_str in artifact_paths: + content, err = await asyncio.to_thread( + _read_artifact, runtime, path_str + ) + if content is not None: + fmt = _detect_artifact_format(path_str) + if fmt == 'jsonl': + # 只取前 50 行，避免超大 + content = '\n'.join(content.splitlines()[:50]) + payload = { + 'step': step, + 'run_dic': run_dic, + 'format': fmt, + 'content': content, + 'source_path': path_str, + 'fetched_at_ms': int(time.time() * 1000), + } + return Response( + content=json.dumps(payload, ensure_ascii=False), + media_type='application/json', + headers={'Cache-Control': 'no-store'}, + ) + if err: + last_error = err + + payload = { + 'step': step, + 'run_dic': run_dic, + 'format': 'empty', + 'content': '', + 'source_path': artifact_paths[0] if artifact_paths else None, + 'fetched_at_ms': int(time.time() * 1000), + 'error': last_error or '产物文件不存在或读取失败', + 'tried_paths': artifact_paths, + } + return Response( + content=json.dumps(payload, ensure_ascii=False), + media_type='application/json', + headers={'Cache-Control': 'no-store'}, + ) + + @app.get('/api/training/pipeline') + async def get_training_pipeline( + session_id: str | None = None, + account_id: str | None = None, + ) -> Response: + sessions_dir = state.account_paths(account_id)['sessions'] + state_path = _session_state_path(sessions_dir, session_id) + state_entries, _ = _read_program_state(state_path) + payload = _hardcoded_autoresearch_pipeline( + session_id, + sessions_dir=sessions_dir, + state_entries=state_entries, + model_config=state.model_config_for(account_id), + ) + _apply_program_state(payload, state_entries) + return Response( + content=json.dumps(payload, ensure_ascii=False), + media_type='application/json', + headers={'Cache-Control': 'no-store'}, + ) + + @app.get('/api/training/pipeline/stream') + async def stream_training_pipeline( + request: Request, + session_id: str | None = None, + account_id: str | None = None, + ) -> StreamingResponse: + sessions_dir = state.account_paths(account_id)['sessions'] + state_path = _session_state_path(sessions_dir, session_id) + + async def event_gen(): + last_signature: str | None = None + last_heartbeat = time.monotonic() + while True: + if await request.is_disconnected(): + break + state_entries, state_mtime = _read_program_state(state_path) + payload = _hardcoded_autoresearch_pipeline( + session_id, + sessions_dir=sessions_dir, + state_entries=state_entries, + ) + _apply_program_state(payload, state_entries) + payload_str = json.dumps(payload, ensure_ascii=False) + signature = f'{state_mtime}|{hash(payload_str)}' + if signature != last_signature: + last_signature = signature + yield f'data: {payload_str}\n\n' + now = time.monotonic() + if now - last_heartbeat > 15: + yield ': heartbeat\n\n' + last_heartbeat = now + await asyncio.sleep(0.5) + + return StreamingResponse( + event_gen(), + media_type='text/event-stream', + headers={ + 'Cache-Control': 'no-store', + 'X-Accel-Buffering': 'no', + 'Connection': 'keep-alive', + }, + ) @app.get('/api/context-budget') async def get_context_budget( @@ -2100,6 +2264,7 @@ def _serialize_stored_session(stored: StoredAgentSession) -> dict[str, Any]: 'usage': stored.usage, 'total_cost_usd': stored.total_cost_usd, 'model': stored.model_config.get('model'), + 'is_training': stored.is_training, } @@ -4091,6 +4256,1289 @@ def _update_session_title(directory: Path, session_id: str, title: str) -> bool: return True +def _update_session_training(directory: Path, session_id: str, is_training: bool) -> bool: + path = _session_json_path(directory, session_id) + if not path.exists(): + return False + try: + data = json.loads(path.read_text(encoding='utf-8')) + except (OSError, json.JSONDecodeError): + return False + data['is_training'] = is_training + _write_session_metadata(path, data) + return True + + +# --------------------------------------------------------------------------- +# Pipeline state watcher: agent declares 「watch」 entries in program-state.jsonl; +# a backend scanner picks them up and polls the target file/dir on its own, +# writing complete/failed entries when the watcher resolves. This decouples UI +# state from agent's discipline (or lack thereof). +# --------------------------------------------------------------------------- + + +def _now_iso_local() -> str: + from datetime import datetime, timezone + return datetime.now(timezone.utc).astimezone().isoformat(timespec='seconds') + + +def _watcher_append(path: Path, entry: dict[str, Any]) -> None: + try: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open('a', encoding='utf-8') as fp: + fp.write(json.dumps(entry, ensure_ascii=False) + '\n') + except OSError: + return + + +class _WatcherManager: + def __init__(self) -> None: + self._tasks: dict[tuple[str, str], asyncio.Task[None]] = {} + + def is_watching(self, session_id: str, step: str) -> bool: + return (session_id, step) in self._tasks + + def register_file_exists( + self, + *, + session_id: str, + state_path: Path, + step: str, + target: Path, + interval: float, + timeout: float, + run_id: str, + agent_state: 'AgentState | None' = None, + account_id: str | None = None, + ) -> None: + key = (session_id, step) + if key in self._tasks: + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + task = loop.create_task( + self._poll_file_exists( + key=key, + state_path=state_path, + step=step, + target=target, + interval=max(2.0, interval), + timeout=max(60.0, timeout), + run_id=run_id, + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + ) + self._tasks[key] = task + + async def _poll_file_exists( + self, + *, + key: tuple[str, str], + state_path: Path, + step: str, + target: Path, + interval: float, + timeout: float, + run_id: str, + agent_state: 'AgentState | None' = None, + account_id: str | None = None, + session_id: str | None = None, + ) -> None: + try: + mode = 'remote' if ( + agent_state is not None + and account_id is not None + and session_id is not None + and _jupyter_runtime_for_session(agent_state, account_id, session_id) + is not None + ) else 'local' + await self._write_log( + state_path, + run_id, + f'watcher 启动 step={step} target={target.name} 周期={int(interval)}s 超时={int(timeout)}s mode={mode}', + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + start = time.monotonic() + while True: + exists = await self._check_target_exists( + target, + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + if exists: + await self._write_step( + state_path, + step=step, + status='complete', + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + await self._write_log( + state_path, + run_id, + f'✅ watcher 检测到 {target.name} 落盘 → step {step} complete', + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + return + if time.monotonic() - start > timeout: + await self._write_step( + state_path, + step=step, + status='failed', + error=f'watcher 超时 {int(timeout)}s, target 未出现: {target}', + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + await self._write_log( + state_path, + run_id, + f'⚠️ watcher 超时 step={step} 未在 {int(timeout)}s 内看到 {target.name}', + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + return + await asyncio.sleep(interval) + except asyncio.CancelledError: + return + except Exception as exc: # noqa: BLE001 + await self._write_log( + state_path, + run_id, + f'⚠️ watcher 异常 step={step}: {exc}', + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + finally: + self._tasks.pop(key, None) + + async def _write_step( + self, + state_path: Path, + *, + step: str, + status: str, + error: str | None = None, + agent_state: 'AgentState | None' = None, + account_id: str | None = None, + session_id: str | None = None, + ) -> None: + entry: dict[str, Any] = { + 'step': step, + 'status': status, + 'ts': _now_iso_local(), + } + if error: + entry['error'] = error + await self._append_state_entry( + state_path, + entry, + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + + async def _write_log( + self, + state_path: Path, + run_id: str, + text: str, + *, + agent_state: 'AgentState | None' = None, + account_id: str | None = None, + session_id: str | None = None, + ) -> None: + entry = { + 'log': { + 'ts': time.strftime('%H:%M:%S'), + 'iter': run_id, + 'text': text, + } + } + await self._append_state_entry( + state_path, + entry, + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + + def cancel_all(self) -> None: + for task in self._tasks.values(): + task.cancel() + self._tasks.clear() + + async def _append_state_entry( + self, + state_path: Path, + entry: dict[str, Any], + *, + agent_state: 'AgentState | None' = None, + account_id: str | None = None, + session_id: str | None = None, + ) -> None: + """Append a state entry. Prefers remote (so sync doesn't overwrite our + writes); falls back to local when no Jupyter binding exists.""" + line = json.dumps(entry, ensure_ascii=False) + '\n' + if ( + agent_state is not None + and account_id is not None + and session_id is not None + ): + runtime = _jupyter_runtime_for_session( + agent_state, account_id, session_id + ) + if runtime is not None: + remote_dir = f'{runtime.binding.workspace_cwd}/output' + remote_path = f'{remote_dir}/program-state.jsonl' + cmd = ( + f'mkdir -p {shlex.quote(remote_dir)} && ' + f'printf %s {shlex.quote(line)} >> {shlex.quote(remote_path)}' + ) + try: + await asyncio.to_thread( + runtime.run_command, + cmd, + timeout_seconds=10.0, + max_output_chars=64, + ) + return + except Exception: # noqa: BLE001 + pass + # Local fallback + await asyncio.to_thread(_watcher_append, state_path, entry) + + async def _check_target_exists( + self, + target: Path, + *, + agent_state: 'AgentState | None' = None, + account_id: str | None = None, + session_id: str | None = None, + ) -> bool: + """File-existence check that prefers remote Jupyter when bound (so + targets on /mnt/* mounts visible to the remote workspace are reachable).""" + if ( + agent_state is not None + and account_id is not None + and session_id is not None + ): + runtime = _jupyter_runtime_for_session( + agent_state, account_id, session_id + ) + if runtime is not None: + cmd = ( + f'test -f {shlex.quote(str(target))} ' + f'&& echo OK || echo MISSING' + ) + try: + result = await asyncio.to_thread( + runtime.run_command, + cmd, + timeout_seconds=10.0, + max_output_chars=64, + ) + except Exception: # noqa: BLE001 + return False + return bool(result and result.stdout.strip().endswith('OK')) + # Fallback: local file system + try: + return target.is_file() + except OSError: + return False + + @staticmethod + def _append_step(state_path: Path, **fields: Any) -> None: + entry: dict[str, Any] = {**fields, 'ts': _now_iso_local()} + _watcher_append(state_path, entry) + + @staticmethod + def _append_log(state_path: Path, run_id: str, text: str) -> None: + entry = { + 'log': { + 'ts': time.strftime('%H:%M:%S'), + 'iter': run_id, + 'text': text, + } + } + _watcher_append(state_path, entry) + + +_watcher_manager = _WatcherManager() + + +def _sync_remote_program_state( + state: AgentState, + account_id: str, + session_id: str, + local_state_path: Path, +) -> None: + """If a Jupyter binding exists, cat the remote program-state.jsonl into local. + Backend reads only the local file; this keeps both in sync so agent writes + on remote (where its bash runs) are visible to UI/watcher logic.""" + runtime = _jupyter_runtime_for_session(state, account_id, session_id) + if runtime is None: + return + remote_state_path = ( + f'{runtime.binding.workspace_cwd}/output/program-state.jsonl' + ) + try: + result = runtime.run_command( + f'test -f {shlex.quote(remote_state_path)} && ' + f'cat {shlex.quote(remote_state_path)} || true', + timeout_seconds=10.0, + max_output_chars=200_000, + ) + except Exception: # noqa: BLE001 + return + if not result or result.exit_code != 0: + return + new_content = result.stdout or '' + if not new_content.strip(): + return + # Skip rewrite if local already matches remote (avoid touching mtime, which + # SSE uses to detect changes). + try: + if local_state_path.is_file() and ( + local_state_path.read_text(encoding='utf-8') == new_content + ): + return + except OSError: + pass + try: + local_state_path.parent.mkdir(parents=True, exist_ok=True) + tmp = local_state_path.with_suffix('.jsonl.sync-tmp') + tmp.write_text(new_content, encoding='utf-8') + tmp.replace(local_state_path) + except OSError: + return + + +def _scan_for_watchers(state: AgentState) -> None: + """Synchronous scan body — used as fallback when no event loop is available. + Prefer _scan_for_watchers_async in normal operation so blocking calls + (remote sync, LLM) run in worker threads instead of the loop.""" + accounts_root = state.session_directory.parent / 'accounts' + if not accounts_root.is_dir(): + return + for account_dir in accounts_root.iterdir(): + if not account_dir.is_dir(): + continue + account_id = account_dir.name + sessions_dir = account_dir / 'sessions' + if not sessions_dir.is_dir(): + continue + for session_dir in sessions_dir.iterdir(): + session_id = session_dir.name + state_path = session_dir / 'output' / 'program-state.jsonl' + try: + _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, + ) + if not state_path.is_file(): + continue + entries, _ = _read_program_state(state_path) + last_status_per_step: dict[str, str] = {} + for entry in entries: + step = entry.get('step') + status = entry.get('status') + if isinstance(step, str) and step and isinstance(status, str): + last_status_per_step[step] = status + for entry in entries: + watch = entry.get('watch') + if not isinstance(watch, dict): + continue + step = watch.get('step') + if not isinstance(step, str) or not step: + continue + if last_status_per_step.get(step) in { + 'complete', + 'failed', + 'cancelled', + }: + continue + if _watcher_manager.is_watching(session_id, step): + continue + kind = watch.get('kind', 'file_exists') + if kind != 'file_exists': + continue + path_str = watch.get('path') + if not isinstance(path_str, str) or not path_str: + continue + try: + interval = float(watch.get('interval', 30) or 30) + timeout = float(watch.get('timeout', 3600) or 3600) + except (TypeError, ValueError): + interval = 30.0 + timeout = 3600.0 + run_id = str(watch.get('run_id') or '?') + _watcher_manager.register_file_exists( + session_id=session_id, + state_path=state_path, + step=step, + target=Path(path_str), + interval=interval, + timeout=timeout, + run_id=run_id, + agent_state=state, + account_id=account_id, + ) + + +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.""" + accounts_root = state.session_directory.parent / 'accounts' + if not accounts_root.is_dir(): + return + try: + account_dirs = list(accounts_root.iterdir()) + except OSError: + return + for account_dir in account_dirs: + if not account_dir.is_dir(): + continue + account_id = account_dir.name + sessions_dir = account_dir / 'sessions' + if not sessions_dir.is_dir(): + continue + try: + session_dirs = list(sessions_dir.iterdir()) + except OSError: + continue + 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, + ) + if not state_path.is_file(): + continue + try: + entries, _ = await asyncio.to_thread( + _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: + step = entry.get('step') + status = entry.get('status') + if isinstance(step, str) and step and isinstance(status, str): + last_status_per_step[step] = status + for entry in entries: + watch = entry.get('watch') + if not isinstance(watch, dict): + continue + step = watch.get('step') + if not isinstance(step, str) or not step: + continue + if last_status_per_step.get(step) in { + 'complete', + 'failed', + 'cancelled', + }: + continue + if _watcher_manager.is_watching(session_id, step): + continue + kind = watch.get('kind', 'file_exists') + if kind != 'file_exists': + continue + path_str = watch.get('path') + if not isinstance(path_str, str) or not path_str: + continue + try: + interval_s = float(watch.get('interval', 30) or 30) + timeout_s = float(watch.get('timeout', 3600) or 3600) + except (TypeError, ValueError): + interval_s = 30.0 + timeout_s = 3600.0 + run_id = str(watch.get('run_id') or '?') + _watcher_manager.register_file_exists( + session_id=session_id, + state_path=state_path, + step=step, + target=Path(path_str), + interval=interval_s, + timeout=timeout_s, + run_id=run_id, + agent_state=state, + account_id=account_id, + ) + + +async def _watcher_scanner_loop( + state: AgentState, interval: float = 5.0 +) -> None: + """Periodically scan all session program-state.jsonl files for new watch + declarations and spawn watchers for them. Blocking calls run in threads.""" + while True: + try: + await _scan_for_watchers_async(state) + except asyncio.CancelledError: + return + except Exception as exc: # noqa: BLE001 + print(f'[watcher-scanner] scan error: {exc}', flush=True) + try: + await asyncio.sleep(interval) + except asyncio.CancelledError: + return + + +def _session_state_path(sessions_dir: Path, session_id: str | None) -> Path | None: + """Resolve //output/program-state.jsonl. Returns None if unsafe id.""" + safe_id = _safe_session_id(session_id) + if safe_id is None: + return None + return sessions_dir / safe_id / 'output' / 'program-state.jsonl' + + +def _read_program_state(path: Path | None) -> tuple[list[dict[str, Any]], float | None]: + """Returns (entries, mtime_seconds). Empty if path is None / missing / unreadable.""" + if path is None: + return [], None + try: + if not path.is_file(): + return [], None + st = path.stat() + text = path.read_text(encoding='utf-8', errors='replace') + except (FileNotFoundError, OSError, PermissionError): + return [], None + entries: list[dict[str, Any]] = [] + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith('#'): + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + entries.append(obj) + return entries, st.st_mtime + + +def _normalize_step_token(value: str) -> str: + return re.sub(r'[\s_\-/.]+', '', value.strip().lower()) + + +def _step_matches_card(step: str, card: dict[str, Any]) -> bool: + s = _normalize_step_token(step) + if not s: + return False + key = _normalize_step_token(str(card.get('key') or '')) + if s == key: + return True + title = _normalize_step_token(str(card.get('title') or '')) + if title and (s in title or title.startswith(s) or s in key): + return True + return False + + +def _apply_program_state( + payload: dict[str, Any], + state_entries: list[dict[str, Any]], +) -> None: + """Mutate payload in place. State entries override card status/progress and + append log lines. KPI entries (legacy `{kpi:..,value:..}`) are ignored — + KPIs are computed by the backend in _compute_kpis.""" + if not state_entries: + return + latest_by_step: dict[str, dict[str, Any]] = {} + appended_logs: list[dict[str, Any]] = [] + for entry in state_entries: + if entry.get('kpi'): + # Backend now owns KPI rendering — explicitly ignore agent-written + # kpi entries to avoid drift between displayed value and authoritative + # source (config.yaml / iteration_log / metric_diff). + continue + if entry.get('log'): + log_obj = entry.get('log') + if isinstance(log_obj, dict): + appended_logs.append(log_obj) + elif isinstance(log_obj, str): + appended_logs.append({'text': log_obj, 'ts': entry.get('ts', '')}) + continue + step = entry.get('step') + if isinstance(step, str) and step: + latest_by_step[step] = entry + + if appended_logs: + existing = payload.setdefault('logs', []) + existing.extend(appended_logs) + + if not latest_by_step: + # logs may still have been applied; status derivation runs below. + latest_by_step = {} + + for step, entry in latest_by_step.items(): + applied = False + for phase in payload.get('phases', []): + for card in phase.get('cards', []): + if _step_matches_card(step, card): + status = entry.get('status') + if isinstance(status, str) and status: + card['status'] = status + if 'progress' in entry: + try: + value = float(entry['progress']) + if value > 1: + value /= 100 + card['progress'] = max(0.0, min(1.0, value)) + except (TypeError, ValueError): + pass + applied = True + break + if applied: + break + + # Re-derive phase tones and overall state. + # 规则：卡片有任何 "已开始过"（running 或 complete）的活动 → 阶段/整体 + # 都视为 running（迭代进行中）。只有全部 pending 才是 idle，全部 complete + # 才是真的完成。 + for phase in payload.get('phases', []): + statuses = [c.get('status') for c in phase.get('cards', [])] + if not statuses: + phase['tone'] = 'pending' + elif all(s == 'complete' for s in statuses): + phase['tone'] = 'complete' + elif 'running' in statuses or 'complete' in statuses: + phase['tone'] = 'running' + elif 'failed' in statuses: + phase['tone'] = 'pending' + else: + phase['tone'] = 'pending' + + all_cards = [ + c for phase in payload.get('phases', []) for c in phase.get('cards', []) + ] + if not all_cards: + return + statuses = [c.get('status') for c in all_cards] + if all(s == 'complete' for s in statuses): + payload['status']['state'] = 'complete' + elif 'running' in statuses or 'complete' in statuses: + # 有任意进展（在跑或完成过任一步）→ 整体迭代是运行中 + payload['status']['state'] = 'running' + elif 'failed' in statuses: + payload['status']['state'] = 'failed' + else: + payload['status']['state'] = 'pending' +def _autoresearch_root() -> Path: + return Path(os.environ.get('AUTORESEARCH_ROOT', '/mnt/wangsenhao/autoresearch-zk')) + + +def _run_history_dir() -> Path: + return Path( + os.environ.get( + 'RUN_HISTORY_DIR', + '/mnt/xiaoai-zk-model-train-tj5/workflow5', + ) + ) + + +def _skill_config_yaml_path() -> Path: + # backend/api/server.py → ../../skills/model-iteration/assets/config.yaml + return ( + Path(__file__).resolve().parent.parent.parent + / 'skills' + / 'model-iteration' + / 'assets' + / 'config.yaml' + ) + + +def _read_workflow_version() -> str | None: + path = _skill_config_yaml_path() + try: + text = path.read_text(encoding='utf-8') + except (FileNotFoundError, OSError): + return None + # Tiny parser to avoid pulling pyyaml just for two lines. + in_cml = False + for raw in text.splitlines(): + stripped = raw.rstrip() + if not stripped or stripped.lstrip().startswith('#'): + continue + if not stripped.startswith(' '): + in_cml = stripped.split(':', 1)[0].strip() == 'cml_eval' + continue + if in_cml: + inner = stripped.strip() + if inner.startswith('version:'): + value = inner.split(':', 1)[1].strip().strip('"\'') + return value or None + return None + + +def _scan_run_history_max() -> int | None: + root = _run_history_dir() + try: + if not root.is_dir(): + return None + except OSError: + return None + best = -1 + for entry in root.iterdir(): + if not entry.is_dir(): + continue + m = re.match(r'^workflow(\d+)$', entry.name) + if m: + try: + value = int(m.group(1)) + except ValueError: + continue + if value > best: + best = value + return best if best >= 0 else None + + +def _read_iteration_log_count() -> int | None: + path = _autoresearch_root() / 'results' / 'iteration_log.jsonl' + try: + if not path.is_file(): + return None + with path.open('r', encoding='utf-8') as fp: + return sum(1 for line in fp if line.strip()) + except (OSError, UnicodeDecodeError): + return None + + +def _extract_trigger_from_state(state_entries: list[dict[str, Any]]) -> str | None: + """Backwards compat: if agent wrote {kpi:'TRIGGER',value:...}, use that. + Backend now owns KPI rendering, but TRIGGER is a passthrough of the user's + own phrasing — fine to read from state.""" + last: str | None = None + for entry in state_entries: + if entry.get('kpi') == 'TRIGGER': + value = entry.get('value') + if isinstance(value, str) and value.strip(): + last = value + return last + + +def _extract_trigger_from_session( + sessions_dir: Path, session_id: str | None +) -> str | None: + """Return the user's first non-empty message that looks like a training + trigger. Used as raw input for _parse_target_set.""" + if not session_id: + return None + safe_id = _safe_session_id(session_id) + if safe_id is None: + return None + path = sessions_dir / safe_id / 'session.json' + try: + data = json.loads(path.read_text(encoding='utf-8')) + except (OSError, json.JSONDecodeError): + return None + messages = data.get('messages') + if not isinstance(messages, list): + return None + for msg in messages: + if not isinstance(msg, dict): + continue + if msg.get('role') != 'user': + continue + content = msg.get('content') + text = content if isinstance(content, str) else '' + if not text: + continue + # Strip blocks etc. + cleaned = re.sub(r'<[^>]+>', ' ', text).strip() + if not cleaned: + continue + if _looks_like_training_trigger(cleaned): + return cleaned + return None + + +def _looks_like_training_trigger(text: str) -> bool: + keywords = ( + '训练', + 'training', + '评测', + '迭代', + '开始', + '需求集合', + '目标集合', + '目标是', + '针对', + 'cml', + '复杂导航', + '.csv', + ) + lower = text.lower() + return any(k.lower() in lower for k in keywords) + + +_TARGET_SET_LLM_CACHE: dict[str, str | None] = {} + + +def _llm_extract_target_set( + model_config: ModelConfig | None, text: str +) -> str | None: + """Use the configured LLM to extract a target set name from a user message. + Cached by message hash so the same input doesn't repeatedly call the model.""" + if not text or not model_config: + return None + text = text.strip() + if len(text) > 4000: + text = text[:4000] + cache_key = hashlib.sha256(text.encode('utf-8')).hexdigest() + if cache_key in _TARGET_SET_LLM_CACHE: + return _TARGET_SET_LLM_CACHE[cache_key] + + # Tighten timeout for this short auxiliary call so SSE doesn't stall. + cfg = replace(model_config, timeout_seconds=min(20.0, model_config.timeout_seconds)) + try: + from src.openai_compat import OpenAICompatClient, OpenAICompatError + except ImportError: + _TARGET_SET_LLM_CACHE[cache_key] = None + return None + + client = OpenAICompatClient(cfg) + system_prompt = ( + '你是一个文本抽取器。任务：从用户的训练触发消息里抽出"目标需求集合"的名称。\n' + '- 名称可能是 CSV 文件名、目录名、或一个标识词\n' + '- 只返回名称字符串本身，不要任何前后缀、引号、说明、标点\n' + '- 没有明确目标时返回大写 NONE' + ) + few_shot = [ + {'role': 'user', 'content': '开始, icl_test'}, + {'role': 'assistant', 'content': 'icl_test'}, + {'role': 'user', 'content': '开始，复杂导航过召专项0511.csv'}, + {'role': 'assistant', 'content': '复杂导航过召专项0511.csv'}, + { + 'role': 'user', + 'content': '我要进行模型训练，目标是icl_test中的复杂导航过召专项0511.csv，基模在/mnt/zhangzhaowen/icl/v2/test/', + }, + {'role': 'assistant', 'content': '复杂导航过召专项0511.csv'}, + {'role': 'user', 'content': '针对 dapan_test 跑一轮 SFT'}, + {'role': 'assistant', 'content': 'dapan_test'}, + {'role': 'user', 'content': '你好'}, + {'role': 'assistant', 'content': 'NONE'}, + ] + messages: list[dict[str, Any]] = [{'role': 'system', 'content': system_prompt}] + messages.extend(few_shot) + messages.append({'role': 'user', 'content': text}) + + try: + result = client.complete(messages, tools=[]) + except OpenAICompatError: + _TARGET_SET_LLM_CACHE[cache_key] = None + return None + except Exception: # noqa: BLE001 — keep KPI render robust to upstream LLM hiccups + _TARGET_SET_LLM_CACHE[cache_key] = None + return None + + answer = (getattr(result, 'content', '') or '').strip().strip('"\'`，,。()（） ') + if not answer or answer.upper() == 'NONE' or len(answer) > 200: + _TARGET_SET_LLM_CACHE[cache_key] = None + return None + # Single-line: take only the first line in case model added explanation + answer = answer.split('\n', 1)[0].strip() + if not answer or answer.upper() == 'NONE': + _TARGET_SET_LLM_CACHE[cache_key] = None + return None + _TARGET_SET_LLM_CACHE[cache_key] = answer + return answer + + +def _parse_target_set( + trigger: str | None, model_config: ModelConfig | None = None +) -> str | None: + """Sync, non-blocking. Reads cache only — never makes the LLM call inline. + Cache is filled by the watcher scanner background loop via + _refresh_target_set_cache_blocking (which is run in a thread).""" + if not trigger: + return None + text = trigger.strip() + if not text: + return None + if len(text) > 4000: + text = text[:4000] + cache_key = hashlib.sha256(text.encode('utf-8')).hexdigest() + return _TARGET_SET_LLM_CACHE.get(cache_key) + + +def _refresh_target_set_cache_blocking( + model_config: ModelConfig | None, trigger: str +) -> None: + """Synchronous wrapper that calls the LLM and populates the cache. + Designed to be invoked via asyncio.to_thread from a background task.""" + if not model_config or not trigger: + return + text = trigger.strip() + if not text: + return + if len(text) > 4000: + text = text[:4000] + cache_key = hashlib.sha256(text.encode('utf-8')).hexdigest() + if cache_key in _TARGET_SET_LLM_CACHE: + return + _llm_extract_target_set(model_config, text) + + +def _read_workflow_metric_diff(run_id: int) -> dict[str, Any] | None: + path = ( + _run_history_dir() + / f'workflow{run_id}' + / 'metric_diff' + / 'lark_template.json' + ) + try: + if not path.is_file(): + return None + return json.loads(path.read_text(encoding='utf-8')) + except (OSError, json.JSONDecodeError): + return None + + +def _format_metric_value(metrics: dict[str, Any] | None, key: str) -> str: + if not metrics: + return '—' + value = metrics.get(key) + if value is None: + return '—' + if isinstance(value, (int, float)): + if 0 <= value <= 1: + return f'{value * 100:.2f}%' + return f'{value:.2f}' + return str(value) + + +def _compute_step0_start_seconds( + state_entries: list[dict[str, Any]], +) -> float | None: + """Earliest cml running ts seen in state file.""" + earliest: float | None = None + for entry in state_entries: + if entry.get('step') != 'cml': + continue + if entry.get('status') != 'running': + continue + ts = entry.get('ts') + if not isinstance(ts, str): + continue + try: + from datetime import datetime + dt = datetime.fromisoformat(ts.replace('Z', '+00:00')) + seconds = dt.timestamp() + except (ValueError, OSError): + continue + if earliest is None or seconds < earliest: + earliest = seconds + return earliest + + +def _format_elapsed(start_seconds: float | None) -> str: + if start_seconds is None: + return '00:00:00' + delta = max(0.0, time.time() - start_seconds) + total = int(delta) + h, rem = divmod(total, 3600) + m, s = divmod(rem, 60) + return f'{h:02d}:{m:02d}:{s:02d}' + + +def _compute_kpis( + sessions_dir: Path, + session_id: str | None, + state_entries: list[dict[str, Any]], + model_config: ModelConfig | None = None, +) -> list[dict[str, Any]]: + """Compute the 7 top KPIs from authoritative sources. Falls back to '—'.""" + run_max = _scan_run_history_max() + run_id_value = str(run_max) if run_max is not None else '—' + version = _read_workflow_version() or '—' + trigger = _extract_trigger_from_state(state_entries) or _extract_trigger_from_session( + sessions_dir, session_id + ) + target_set = _parse_target_set(trigger, model_config=model_config) or '—' + metrics = _read_workflow_metric_diff(run_max) if run_max is not None else None + target_metric = _format_metric_value(metrics, 'target_set_pass_rate') + overall_metric = _format_metric_value(metrics, 'overall_car_pass_rate') + specific_metric = _format_metric_value(metrics, 'specific_test_pass_rate') + iter_count = _read_iteration_log_count() + if iter_count is None: + iteration = 'R0-baseline' + elif iter_count == 0: + iteration = 'R0-baseline' + else: + iteration = f'R{iter_count}' + elapsed = _format_elapsed(_compute_step0_start_seconds(state_entries)) + target_value = ( + f'{target_set} · {target_metric}' if target_metric != '—' else target_set + ) + return [ + {'label': 'RUN ID', 'value': run_id_value}, + {'label': 'VERSION', 'value': version}, + {'label': '目标集合', 'value': target_value}, + {'label': '大盘车载', 'value': overall_metric}, + {'label': 'SPECIFIC TEST', 'value': specific_metric}, + {'label': 'ITERATION', 'value': iteration}, + {'label': '耗时', 'value': elapsed, 'icon': 'clock'}, + ] + + +def _resolve_run_dic_from_state( + state_entries: list[dict[str, Any]], +) -> int | None: + """Extract runDic from any place agent might leave it in program-state.jsonl: + 1. Top-level `runDic` field on step entries + 2. Any string field containing `workflow` or `runDic=N` (covers + report_path, log.text, watch.path, etc.) + 3. Latest occurrence wins (reverse iteration).""" + workflow_re = re.compile(r'workflow(\d+)') + rundic_re = re.compile(r'runDic\s*[=:取]?\s*(\d+)') + + def _scan_strings(obj: Any) -> int | None: + if isinstance(obj, str): + m = workflow_re.search(obj) + if m: + return int(m.group(1)) + m = rundic_re.search(obj) + if m: + return int(m.group(1)) + elif isinstance(obj, dict): + for v in obj.values(): + hit = _scan_strings(v) + if hit is not None: + return hit + elif isinstance(obj, list): + for v in obj: + hit = _scan_strings(v) + if hit is not None: + return hit + return None + + for entry in reversed(state_entries): + # 1. 显式字段 runDic + run_dic = entry.get('runDic') or entry.get('run_dic') or entry.get('rundic') + if isinstance(run_dic, int): + return run_dic + if isinstance(run_dic, str) and run_dic.isdigit(): + return int(run_dic) + # 2. 扫所有字符串字段（递归） + hit = _scan_strings(entry) + if hit is not None: + return hit + return None + + +def _step_artifact_paths(step: str, run_dic: int | None) -> list[str]: + """Map step key → ordered list of remote artifact paths to try.""" + auto_root = os.environ.get( + 'AUTORESEARCH_ROOT', '/mnt/wangsenhao/autoresearch-zk' + ) + metric_root = os.environ.get( + 'RUN_HISTORY_DIR', '/mnt/xiaoai-zk-model-train-tj5/workflow5' + ) + paths: list[str] = [] + if step == 'cml': + if run_dic is not None: + paths.append( + f'{metric_root}/workflow{run_dic}/metric_diff/lark_template.json' + ) + paths.append( + f'{metric_root}/workflow{run_dic}/metric_diff/specific_comparison.csv' + ) + elif step == 'gold-drift': + if run_dic is not None: + paths.append(f'{auto_root}/results/gold_drift/drift_{run_dic}.json') + elif step in {'dist-analysis', 'report'}: + if run_dic is not None: + paths.append(f'{auto_root}/results/workflow{run_dic}.md') + elif step == 'hypothesis': + paths.append(f'{auto_root}/results/iteration_log.jsonl') + elif step == 'augment': + if run_dic is not None: + paths.append( + f'{auto_root}/results/augment_raw/augment_{run_dic}_raw.jsonl' + ) + paths.append( + f'{auto_root}/ai-planning/data/train_set/zk_intent/augment_{run_dic}.jsonl' + ) + elif step == 'verify': + paths.append(f'{auto_root}/results/iteration_log.jsonl') + elif step == 'sft': + paths.append(f'{auto_root}/results/iteration_log.jsonl') + elif step == 'log': + paths.append(f'{auto_root}/results/iteration_log.jsonl') + paths.append(f'{auto_root}/results/error_registry.jsonl') + elif step == 'next-round': + paths.append(f'{auto_root}/results/iteration_log.jsonl') + return paths + + +def _detect_artifact_format(path: str) -> str: + lower = path.lower() + if lower.endswith('.md'): + return 'markdown' + if lower.endswith('.jsonl'): + return 'jsonl' + if lower.endswith('.json'): + return 'json' + if lower.endswith('.csv'): + return 'csv' + return 'text' + + +def _read_artifact( + runtime: 'JupyterRuntimeSession | None', remote_path: str +) -> tuple[str | None, str | None]: + """Read a remote artifact via Jupyter `cat`. Falls back to local Path read + if no remote runtime (dev / unbound sessions). Returns (content, error).""" + if runtime is not None: + cmd = ( + f'test -f {shlex.quote(remote_path)} && ' + f'cat {shlex.quote(remote_path)} || echo __MISSING__' + ) + try: + result = runtime.run_command( + cmd, + timeout_seconds=15.0, + max_output_chars=400_000, + ) + except Exception as exc: # noqa: BLE001 + return None, f'remote read failed: {exc}' + if not result: + return None, 'no result' + out = result.stdout or '' + # Trim our own marker if file missing + if out.strip().endswith('__MISSING__'): + return None, f'remote file not found: {remote_path}' + return out, None + try: + p = Path(remote_path) + if not p.is_file(): + return None, f'file not found: {remote_path}' + return p.read_text(encoding='utf-8', errors='replace'), None + except OSError as exc: + return None, f'local read failed: {exc}' + + +def _hardcoded_autoresearch_pipeline( + session_id: str | None, + sessions_dir: Path | None = None, + state_entries: list[dict[str, Any]] | None = None, + model_config: ModelConfig | None = None, +) -> dict[str, Any]: + """Canonical autoresearch model-iteration pipeline structure. + + Card / phase / log dynamics still come from program-state.jsonl, but the + 7 top KPIs are now backend-owned (read from skills/config.yaml, + iteration_log, run history dir, CML metric_diff, session state) — agent + cannot override them via state file. + """ + now_ms = int(time.time() * 1000) + if sessions_dir is not None: + kpis = _compute_kpis( + sessions_dir, + session_id, + state_entries or [], + model_config=model_config, + ) + else: + # Defensive default if caller didn't pass sessions_dir. + kpis = [ + {'label': 'RUN ID', 'value': '—'}, + {'label': 'VERSION', 'value': '—'}, + {'label': '目标集合', 'value': '—'}, + {'label': '大盘车载', 'value': '—'}, + {'label': 'SPECIFIC TEST', 'value': '—'}, + {'label': 'ITERATION', 'value': 'R0-baseline'}, + {'label': '耗时', 'value': '00:00:00', 'icon': 'clock'}, + ] + return { + 'session_id': session_id or '', + 'generated_at_ms': now_ms, + 'available': True, + 'status': {'mode': 'Model Iteration', 'state': 'pending'}, + 'kpis': kpis, + 'phases': [ + { + 'key': 'eval', + 'label': '评测与诊断', + 'tone': 'pending', + 'cards': [ + {'key': 'cml', 'icon': 'bar-chart', 'title': 'CML 评测', 'subtitle': 'workflow metric_diff', 'status': 'pending'}, + {'key': 'gold-drift', 'icon': 'microscope', 'title': 'Gold Drift', 'subtitle': 'RAG 检查漂移', 'status': 'pending'}, + {'key': 'dist-analysis', 'icon': 'trending-up', 'title': '分层结果分析', 'subtitle': '需求集合 / 大盘车载 / specific', 'status': 'pending'}, + {'key': 'report', 'icon': 'file-text', 'title': '问题分析 & 报告', 'subtitle': 'workflow.md', 'status': 'pending'}, + ], + }, + { + 'key': 'intervene', + 'label': '假设与干预', + 'tone': 'pending', + 'cards': [ + {'key': 'hypothesis', 'icon': 'dna', 'title': '形成假设', 'subtitle': 'iteration_log hypothesis', 'status': 'pending'}, + {'key': 'augment', 'icon': 'flask', 'title': '数据增强', 'subtitle': 'augment_.jsonl', 'status': 'pending'}, + {'key': 'verify', 'icon': 'shield', 'title': '修改返回验证', 'subtitle': 'sanity + 人审', 'status': 'pending'}, + ], + }, + { + 'key': 'train', + 'label': '训练与回归', + 'tone': 'pending', + 'cards': [ + {'key': 'sft', 'icon': 'graduation-cap', 'title': 'SFT 训练', 'subtitle': 'submit_sft_via_cml.sh', 'status': 'pending'}, + {'key': 'log', 'icon': 'clipboard-list', 'title': '记录迭代日志', 'subtitle': 'iteration_log.jsonl', 'status': 'pending'}, + {'key': 'next-round', 'icon': 'refresh-cw', 'title': '下一轮评测', 'subtitle': '回 Step 0', 'status': 'pending'}, + ], + }, + ], + 'logs': [], + } + + def _session_mtime(path: Path) -> float: try: return path.stat().st_mtime diff --git a/frontend/app/app/api/claw/training/pipeline/route.ts b/frontend/app/app/api/claw/training/pipeline/route.ts new file mode 100644 index 0000000..3bd10e8 --- /dev/null +++ b/frontend/app/app/api/claw/training/pipeline/route.ts @@ -0,0 +1,26 @@ +import { getCurrentAccount } from "@/lib/claw-auth"; + +const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; + +export async function GET(request: Request) { + const account = await getCurrentAccount(); + if (!account) return Response.json({ error: "unauthorized" }, { status: 401 }); + + const incoming = new URL(request.url); + const sessionId = incoming.searchParams.get("session_id") ?? ""; + + const url = new URL(`${CLAW_API_URL}/api/training/pipeline`); + url.searchParams.set("account_id", account.id); + if (sessionId) url.searchParams.set("session_id", sessionId); + + const response = await fetch(url, { cache: "no-store" }); + const payload = await response.text(); + return new Response(payload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + "cache-control": "no-store", + }, + }); +} diff --git a/frontend/app/app/api/claw/training/pipeline/stream/route.ts b/frontend/app/app/api/claw/training/pipeline/stream/route.ts new file mode 100644 index 0000000..5952f3c --- /dev/null +++ b/frontend/app/app/api/claw/training/pipeline/stream/route.ts @@ -0,0 +1,38 @@ +import { getCurrentAccount } from "@/lib/claw-auth"; + +const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; + +export const dynamic = "force-dynamic"; + +export async function GET(request: Request) { + const account = await getCurrentAccount(); + if (!account) { + return new Response("unauthorized", { status: 401 }); + } + + const incoming = new URL(request.url); + const sessionId = incoming.searchParams.get("session_id") ?? ""; + + const url = new URL(`${CLAW_API_URL}/api/training/pipeline/stream`); + url.searchParams.set("account_id", account.id); + if (sessionId) url.searchParams.set("session_id", sessionId); + + const upstream = await fetch(url, { + cache: "no-store", + signal: request.signal, + }); + + if (!upstream.body) { + return new Response("upstream has no body", { status: 502 }); + } + + return new Response(upstream.body, { + status: upstream.status, + headers: { + "content-type": "text/event-stream", + "cache-control": "no-store", + connection: "keep-alive", + "x-accel-buffering": "no", + }, + }); +} diff --git a/frontend/app/app/api/claw/training/step-detail/route.ts b/frontend/app/app/api/claw/training/step-detail/route.ts new file mode 100644 index 0000000..2615893 --- /dev/null +++ b/frontend/app/app/api/claw/training/step-detail/route.ts @@ -0,0 +1,34 @@ +import { getCurrentAccount } from "@/lib/claw-auth"; + +const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; + +export async function GET(request: Request) { + const account = await getCurrentAccount(); + if (!account) return Response.json({ error: "unauthorized" }, { status: 401 }); + + const incoming = new URL(request.url); + const sessionId = incoming.searchParams.get("session_id") ?? ""; + const step = incoming.searchParams.get("step") ?? ""; + if (!sessionId || !step) { + return Response.json( + { error: "session_id and step are required" }, + { status: 400 }, + ); + } + + const url = new URL(`${CLAW_API_URL}/api/training/step-detail`); + url.searchParams.set("account_id", account.id); + url.searchParams.set("session_id", sessionId); + url.searchParams.set("step", step); + + const response = await fetch(url, { cache: "no-store" }); + const payload = await response.text(); + return new Response(payload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + "cache-control": "no-store", + }, + }); +} diff --git a/frontend/app/app/assistant.tsx b/frontend/app/app/assistant.tsx index 20a0f31..903159c 100644 --- a/frontend/app/app/assistant.tsx +++ b/frontend/app/app/assistant.tsx @@ -1,16 +1,18 @@ "use client"; import type { ExportedMessageRepository } from "@assistant-ui/core"; -import { AssistantRuntimeProvider } from "@assistant-ui/react"; +import { AssistantRuntimeProvider, useAuiState } from "@assistant-ui/react"; import { AssistantChatTransport, useChatRuntime, } from "@assistant-ui/react-ai-sdk"; import type { UIMessage } from "ai"; -import { useCallback, useEffect, useMemo, useRef } from "react"; +import { PanelRightOpenIcon, XIcon } from "lucide-react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { ActivityPanel, ActivityProvider, + useActivityPanel, } from "@/components/assistant-ui/activity-panel"; import { Thread } from "@/components/assistant-ui/thread"; import { @@ -18,18 +20,26 @@ import { toReplayRepository, } from "@/components/assistant-ui/thread-list"; import { ThreadListSidebar } from "@/components/assistant-ui/threadlist-sidebar"; +import { TrainingPipelinePanel } from "@/components/training/training-pipeline-panel"; +import { Button } from "@/components/ui/button"; import { SidebarInset, SidebarProvider, SidebarTrigger, } from "@/components/ui/sidebar"; import { + ACTIVE_SESSION_CHANGED_EVENT, readActiveSessionId, readPendingWorkspaceSessionId, writeActiveSessionId, } from "@/lib/claw-active-session"; import { ClawSessionReplayProvider } from "@/lib/claw-session-replay"; import { pushSessionUrl } from "@/lib/claw-session-url"; +import { cn } from "@/lib/utils"; +import { + useAppliedTraining, + useModelIterationEnabled, +} from "@/lib/use-training-mode"; import { ClawAccountGate, useClawAccount } from "./claw-account-gate"; type AssistantProps = { @@ -153,12 +163,7 @@ export const Assistant = ({ initialSessionId }: AssistantProps = {}) => { onLogout={() => setAccount(null)} /> -
-
- -
- -
+ @@ -168,6 +173,190 @@ export const Assistant = ({ initialSessionId }: AssistantProps = {}) => { ); }; +// 用户在对话里表达「我要开始/进行模型训练（这一轮迭代）」时命中的关键短语。 +// 要求动词或专属触发词必须存在；避免 "什么是模型训练" 之类无意义匹配。 +const APPLY_INTENT_PATTERNS: RegExp[] = [ + // 触发短语：开始，xxx / 开始, xxx + /(?:^|\s|[，,。.!?])开始\s*[，,]\s*\S+/, + // 单纯训练动作：进行/开始/启动/开启/执行/发起/做 + (模型) + 训练 + /(?:进行|开始|启动|开启|执行|发起|做)\s*(?:模型)?\s*训练/i, + // 训练 + 目标集合 + /训练[\s\S]{0,60}?(?:目标集合|需求集合|specific[\s_-]?test|集合)/i, + // 跟着 program.md 的旧表达保留 + /(?:应用|加载|载入|启用|装载|使用|跑|按照?|根据|用)\s*\S*program\.?md/i, + /program\.?md[\s\S]{0,120}?(?:训练|应用|启用|跑起来|跑一下)/i, + /apply\s+program(?:\.md)?/i, + /load\s+program(?:\.md)?/i, + /(?:start|begin|run)\s+(?:the\s+)?training/i, +]; + +function detectApplyIntent(text: string): boolean { + return APPLY_INTENT_PATTERNS.some((re) => re.test(text)); +} + +function AssistantWorkspace() { + const [activeSessionId, setActiveSessionId] = useState(() => + readActiveSessionId(), + ); + const skill = useModelIterationEnabled(); + const { applied, setApplied } = useAppliedTraining(activeSessionId); + const { close: closeActivityPanel } = useActivityPanel(); + const showPipeline = skill.enabled && applied; + const messages = useAuiState((s) => s.thread.messages); + const lastProcessedUserMsgId = useRef(null); + + useEffect(() => { + const update = () => setActiveSessionId(readActiveSessionId()); + const handleChanged = (event: Event) => { + const detail = (event as CustomEvent<{ sessionId?: string | null }>) + .detail; + setActiveSessionId(detail?.sessionId ?? readActiveSessionId()); + }; + window.addEventListener(ACTIVE_SESSION_CHANGED_EVENT, handleChanged); + window.addEventListener("focus", update); + update(); + return () => { + window.removeEventListener(ACTIVE_SESSION_CHANGED_EVENT, handleChanged); + window.removeEventListener("focus", update); + }; + }, []); + + useEffect(() => { + closeActivityPanel(); + }, [showPipeline, closeActivityPanel]); + + // Skill 关掉时撤销 applied，避免重新启用就立刻弹开。 + useEffect(() => { + if (!skill.enabled && applied) { + setApplied(false); + } + }, [skill.enabled, applied, setApplied]); + + // 自动应用：skill 启用 + 当前 session 后端的 pipeline 已经有进展（任何 + // 卡 running 或 complete）→ applied=true。这样关 tab 重开 / 刷新 / 切回 + // 已在跑的会话都能自动恢复 monitor 面板，不依赖触发词。 + useEffect(() => { + if (!skill.enabled) return; + if (applied) return; + if (!activeSessionId) return; + let cancelled = false; + const url = `/api/claw/training/pipeline?session_id=${encodeURIComponent(activeSessionId)}`; + fetch(url, { cache: "no-store" }) + .then((res) => (res.ok ? res.json() : null)) + .then((payload) => { + if (cancelled || !payload) return; + const phases = Array.isArray(payload.phases) ? payload.phases : []; + const hasProgress = phases.some( + (ph: { cards?: Array<{ status?: string }> }) => + Array.isArray(ph.cards) && + ph.cards.some( + (c) => c.status === "running" || c.status === "complete", + ), + ); + if (hasProgress) setApplied(true); + }) + .catch(() => { + /* ignore */ + }); + return () => { + cancelled = true; + }; + }, [skill.enabled, applied, activeSessionId, setApplied]); + + // 监听用户消息：skill 启用 + 检测到训练触发短语 → applied = true。 + useEffect(() => { + if (!skill.enabled) return; + if (applied) return; + if (!Array.isArray(messages) || messages.length === 0) return; + // 找最新一条 user 消息 + let latestUser: (typeof messages)[number] | null = null; + for (let i = messages.length - 1; i >= 0; i -= 1) { + const m = messages[i]; + if (m.role === "user") { + latestUser = m; + break; + } + } + if (!latestUser) return; + if (lastProcessedUserMsgId.current === latestUser.id) return; + lastProcessedUserMsgId.current = latestUser.id; + const text = collectUserMessageText(latestUser); + if (text && detectApplyIntent(text)) { + setApplied(true); + } + }, [messages, skill.enabled, applied, setApplied]); + + // 注意：用同一棵树 + 条件渲染，保证 和 在 + // pipeline 切换前后都保持挂载，否则它们会 remount， + // useJupyterWorkspaceStatus 的内部 state 会被清空，UI 上 Jupyter 工作区 + // indicator 会闪一下变成"切换工作区"。 + return ( +
+ {showPipeline ? ( +
+ + +
+ ) : null} +
+
+ +
+
+ + {showPipeline ? : null} +
+ ); +} + +function collectUserMessageText(message: { content?: unknown }): string { + const content = message.content; + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + const parts: string[] = []; + for (const part of content) { + if (typeof part === "string") { + parts.push(part); + } else if (part && typeof part === "object") { + const text = (part as { text?: unknown }).text; + if (typeof text === "string") parts.push(text); + } + } + return parts.join("\n"); +} + +function ActivityDrawerTrigger() { + const { open, openActivity } = useActivityPanel(); + if (open) return null; + return ( + + ); +} + function getLastSessionId(messages: UIMessage[]) { for (const message of [...messages].reverse()) { if (message.role !== "assistant") continue; diff --git a/frontend/app/app/globals.css b/frontend/app/app/globals.css index 7273c35..5d74f59 100644 --- a/frontend/app/app/globals.css +++ b/frontend/app/app/globals.css @@ -51,6 +51,86 @@ background-position: -100% 0; } } + --animate-pipeline-running: pipeline-running-glow 1800ms ease-in-out infinite; + @keyframes pipeline-running-glow { + 0%, + 100% { + box-shadow: + 0 0 0 1px rgb(52 211 153 / 0.45), + 0 0 0 0 rgb(52 211 153 / 0); + border-color: rgb(52 211 153 / 0.6); + } + 50% { + box-shadow: + 0 0 0 1px rgb(52 211 153 / 0.95), + 0 0 18px 2px rgb(52 211 153 / 0.4); + border-color: rgb(52 211 153 / 1); + } + } + --animate-pipeline-running-blue: pipeline-running-glow-blue 1800ms ease-in-out + infinite; + @keyframes pipeline-running-glow-blue { + 0%, + 100% { + box-shadow: + 0 0 0 1px rgb(56 189 248 / 0.45), + 0 0 0 0 rgb(56 189 248 / 0); + border-color: rgb(56 189 248 / 0.7); + } + 50% { + box-shadow: + 0 0 0 1px rgb(56 189 248 / 0.95), + 0 0 18px 2px rgb(56 189 248 / 0.45); + border-color: rgb(56 189 248 / 1); + } + } + --animate-pipeline-progress-shimmer: pipeline-progress-shimmer 1400ms linear + infinite; + @keyframes pipeline-progress-shimmer { + from { + transform: translateX(-100%); + } + to { + transform: translateX(250%); + } + } + --animate-pipeline-pill-glow: pipeline-pill-glow 1800ms ease-in-out infinite; + @keyframes pipeline-pill-glow { + 0%, + 100% { + box-shadow: + 0 0 0 1px rgb(52 211 153 / 0.3), + 0 0 0 0 rgb(52 211 153 / 0); + border-color: rgb(52 211 153 / 0.45); + } + 50% { + box-shadow: + 0 0 0 1px rgb(52 211 153 / 0.7), + 0 0 10px 2px rgb(52 211 153 / 0.35); + border-color: rgb(52 211 153 / 0.85); + } + } + --animate-pipeline-dot-ping: pipeline-dot-ping 1400ms ease-out infinite; + @keyframes pipeline-dot-ping { + 0% { + transform: scale(0.6); + opacity: 0.9; + } + 80%, + 100% { + transform: scale(2.2); + opacity: 0; + } + } + --animate-icon-spin: icon-spin 3200ms linear infinite; + @keyframes icon-spin { + from { + transform: rotate(0deg); + } + to { + transform: rotate(360deg); + } + } } .doc-page-v2 { diff --git a/frontend/app/components/assistant-ui/activity-panel.tsx b/frontend/app/components/assistant-ui/activity-panel.tsx index 6e748b7..b96df8c 100644 --- a/frontend/app/components/assistant-ui/activity-panel.tsx +++ b/frontend/app/components/assistant-ui/activity-panel.tsx @@ -41,6 +41,7 @@ import { CollapsibleContent, CollapsibleTrigger, } from "@/components/ui/collapsible"; +import { Sheet, SheetContent, SheetTitle } from "@/components/ui/sheet"; import { ACTIVE_SESSION_CHANGED_EVENT, readActiveSessionId, @@ -140,7 +141,7 @@ type ActivityItem = { type LiveRunEvent = NonNullable[number]; -export function ActivityPanel() { +export function ActivityPanel({ asDrawer = false }: { asDrawer?: boolean } = {}) { const { open, mode, @@ -221,20 +222,12 @@ export function ActivityPanel() { prevLatestIdRef.current = latestId; }, [items, mode, openActivity, selectedId]); - if (!open) return null; + if (!asDrawer && !open) return null; - if (mode === "files") { - return ( - - ); - } + const isFilesMode = mode === "files"; - return ( -