From cf5e74d80a39cffe39ede08ba52a506c6af7cb98 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Sun, 12 Jul 2026 12:20:37 +0800 Subject: [PATCH] feat(activity): SRCP v1 step gates, pause/resume/stop control (v1.2.0) Co-authored-by: Cursor --- README.md | 94 ++++++++++- pyproject.toml | 2 +- src/jiangchang_skill_core/__init__.py | 10 ++ src/jiangchang_skill_core/activity.py | 225 ++++++++++++++++++++++++-- tests/test_activity_control.py | 218 +++++++++++++++++++++++++ 5 files changed, 536 insertions(+), 13 deletions(-) create mode 100644 tests/test_activity_control.py diff --git a/README.md b/README.md index a0a1f41..b66be4e 100644 --- a/README.md +++ b/README.md @@ -87,7 +87,10 @@ send_prompt_via_composer(page, "第一行\n第二行") | `emit(text, ...)` | 写 Run Journal（`activity` / `progress` / `warn` / `debug`） | | `step(text)` | `emit(type='activity')` 别名 | | `finish(status=..., **fields)` | 写终态事件 + **唯一**向 stdout 输出单行 result JSON | -| `rpa_step(name)` | 装饰器：进入/成功/失败自动 emit | +| `rpa_step(name)` | 装饰器：闸门 + 进入/成功/失败自动 emit | +| `checkpoint()` | consult control（不写步骤事件） | +| `interruptible_sleep(sec)` | 可暂停/停止的 async sleep | +| `JobStopped` | 宿主 stop 时抛出 | | `job_context(...)` | 可选包裹 main；异常时自动 `finish(failed)` | ### Run Journal 路径（宿主 tail 此文件） @@ -133,6 +136,95 @@ emit("手动步骤说明", skill="publish-toutiao", stage="rpa") `RpaVideoSession.add_step()` 在录屏字幕之外默认同步 `emit(..., stage="rpa.video")`；可通过 `emit_activity=False` 关闭。 +## Skill Runtime Control Protocol (SRCP) v1 — 1.2.0+ + +宿主通过 **control 文件**向技能进程下发暂停 / 继续 / 停止；platform-kit 在**步骤闸门**自动读取，技能作者无需手写控制逻辑。 + +### control.json（宿主写、kit 读） + +``` +{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.control.json +``` + +```json +{ + "schema_version": 1, + "command": "none", + "reason": "user", + "ts": 1730000000000, + "nonce": 1 +} +``` + +| command | 行为 | +|---------|------| +| `none` | 正常运行 | +| `pause` | 下一步闸门进入暂停（阻塞直到 resume 或 stop） | +| `resume` | 从暂停恢复 | +| `stop` | 协作式停止，抛出 `JobStopped` | + +宿主更新命令时应递增 `nonce`（kit 每次闸门重新读文件，不依赖缓存）。 + +### 步骤闸门（自动挂载） + +| 调用点 | 闸门行为 | +|--------|----------| +| `emit()` / `step()` | 进入步骤前 gate + 写 Journal | +| `@rpa_step` | 进入步骤前 gate，再写 ▶ / ✓ | +| `RpaVideoSession.add_step()` | 经 `emit()` 自动 gate | +| `interruptible_sleep()` | 每 200ms 切片 `checkpoint()`（可暂停/停止） | + +暂停在**当前步骤边界**生效（与影刀「当前指令执行完后暂停」一致），不在 DOM 原子操作中途冻结。 + +### lifecycle Journal 事件 + +闸门写入 `type: lifecycle` 事件，供宿主任务中心显示运行 / 暂停 / 继续状态： + +```json +{"type":"lifecycle","status":"paused","text":"已暂停，等待继续","step_index":3,"step_name":"打开页面",...} +``` + +### API + +| API | 作用 | +|-----|------| +| `checkpoint()` | 仅 consult control（长等待循环内使用） | +| `interruptible_sleep(sec)` | async 可中断 sleep | +| `interruptible_sleep_sync(sec)` | sync 可中断 sleep | +| `JobStopped` | 宿主 stop 时抛出；`job_context` 自动 `finish(..., error_code="JOB_STOPPED")` | +| `get_control_path()` | 当前 job 的 control 文件路径（诊断） | + +环境变量： + +- `JIANGCHANG_CONTROL_POLL_SEC` — 暂停轮询间隔（默认 `0.2`） +- `JIANGCHANG_CONTROL_DISABLED=1` — 禁用闸门（仅测试） + +### RPA 示例（拆步 + 可中断等待） + +```python +from jiangchang_skill_core.activity import ( + emit, + finish, + job_context, + interruptible_sleep, + rpa_step, +) + +def cmd_run(...): + with job_context(skill="demo-skill"): + emit("开始任务", skill="demo-skill", stage="run") + + @rpa_step("打开后台") + async def open_admin(page): + await page.goto("https://example.com") + await interruptible_sleep(1.5) + + await open_admin(page) + finish(status="success", message="完成", skill="demo-skill") +``` + +技能作者规范以 **skill-template** 为准；本 README 为实现细节参考。 + ### 兄弟技能 CLI 桥 ```python diff --git a/pyproject.toml b/pyproject.toml index 2ab96ca..b21d9f9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.1.0" +version = "1.2.0" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_skill_core/__init__.py b/src/jiangchang_skill_core/__init__.py index 6449f8d..b86ce2e 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -1,8 +1,13 @@ from .activity import ( + JobStopped, + checkpoint, emit, finish, + get_control_path, get_job_id, get_run_mode, + interruptible_sleep, + interruptible_sleep_sync, job_context, rpa_step, step, @@ -68,14 +73,19 @@ except ImportError: rpa = None # type: ignore[assignment,misc] __all__ = [ + "JobStopped", "call_sibling_json", "call_sibling_json_array", + "checkpoint", "emit", "finish", + "get_control_path", "get_job_id", "get_run_mode", "get_runs_dir", "ensure_runs_dir", + "interruptible_sleep", + "interruptible_sleep_sync", "job_context", "rpa_step", "step", diff --git a/src/jiangchang_skill_core/activity.py b/src/jiangchang_skill_core/activity.py index e9ae27a..60a3718 100644 --- a/src/jiangchang_skill_core/activity.py +++ b/src/jiangchang_skill_core/activity.py @@ -1,8 +1,11 @@ """ -Skill Activity Stream (SAS) v1 — Run Journal + stdout result contract. +Skill Activity Stream (SAS) v1 + Skill Runtime Control Protocol (SRCP) v1. emit() / journal events never touch stdout (Agent/bash tool result stays clean). finish() is the only API that writes a single-line result JSON to stdout. + +SRCP: host writes {job_id}.control.json; kit consults at step gates (emit, +rpa_step, interruptible_sleep) for pause / resume / stop. """ from __future__ import annotations @@ -21,22 +24,47 @@ from typing import Any, Callable, Iterator, Optional, TypeVar from .runtime_env import ensure_runs_dir, get_runs_dir _SCHEMA_VERSION = 1 +_CONTROL_SCHEMA_VERSION = 1 _VALID_EMIT_TYPES = frozenset({"activity", "progress", "warn", "debug"}) +_VALID_CONTROL_COMMANDS = frozenset({"none", "pause", "resume", "stop"}) _RESULT_MAX_CHARS = 4096 +_DEFAULT_CONTROL_POLL_SEC = 0.2 _cached_job_id: Optional[str] = None _journal_path: Optional[str] = None _meta_written: bool = False _journal_file: Optional[Any] = None +_step_index: int = 0 +_current_step_name: Optional[str] = None F = TypeVar("F", bound=Callable[..., Any]) +class JobStopped(Exception): + """Raised when the host requests stop via control.json.""" + + def _activity_debug_enabled() -> bool: v = (os.getenv("JIANGCHANG_ACTIVITY_DEBUG") or "").strip().lower() return v in ("1", "true", "yes", "on") +def _control_disabled() -> bool: + v = (os.getenv("JIANGCHANG_CONTROL_DISABLED") or "").strip().lower() + return v in ("1", "true", "yes", "on") + + +def _control_poll_sec() -> float: + raw = (os.getenv("JIANGCHANG_CONTROL_POLL_SEC") or "").strip() + if not raw: + return _DEFAULT_CONTROL_POLL_SEC + try: + value = float(raw) + except ValueError: + return _DEFAULT_CONTROL_POLL_SEC + return max(0.05, value) + + def get_run_mode() -> str: mode = (os.getenv("JIANGCHANG_RUN_MODE") or "").strip().lower() if mode in ("job", "cli"): @@ -54,6 +82,12 @@ def get_job_id() -> str: return _cached_job_id +def get_control_path() -> str: + """Path to the host-owned control file for the current job.""" + ensure_runs_dir() + return os.path.join(get_runs_dir(), f"{get_job_id()}.control.json") + + def _get_journal_path() -> str: global _journal_path if _journal_path is None: @@ -93,6 +127,103 @@ def _now_ms() -> int: return int(time.time() * 1000) +def _read_control_payload() -> dict[str, Any] | None: + path = get_control_path() + try: + with open(path, encoding="utf-8") as handle: + payload = json.load(handle) + except FileNotFoundError: + return None + except (OSError, json.JSONDecodeError): + return None + if not isinstance(payload, dict): + return None + return payload + + +def _read_control_command() -> str: + payload = _read_control_payload() + if not payload: + return "none" + command = payload.get("command") + if not isinstance(command, str): + return "none" + normalized = command.strip().lower() + if normalized not in _VALID_CONTROL_COMMANDS: + return "none" + return normalized + + +def _write_lifecycle_event( + status: str, + text: str, + *, + step_name: str | None = None, + skill: str | None = None, +) -> None: + event: dict[str, Any] = { + "type": "lifecycle", + "status": status, + "text": text, + "schema_version": _SCHEMA_VERSION, + "control_schema_version": _CONTROL_SCHEMA_VERSION, + "ts": _now_ms(), + "step_index": _step_index, + } + name = step_name or _current_step_name + if name: + event["step_name"] = name + if skill is not None: + event["skill"] = skill + _write_journal_line(event) + + +def _wait_while_paused(*, step_name: str | None = None) -> None: + poll = _control_poll_sec() + while True: + command = _read_control_command() + if command == "stop": + _write_lifecycle_event("stopping", "任务已停止", step_name=step_name) + raise JobStopped("Job stopped by host.") + if command in ("none", "resume"): + _write_lifecycle_event("resumed", "继续执行", step_name=step_name) + return + time.sleep(poll) + + +def _apply_control(*, advance_step: bool, step_name: str | None = None) -> None: + if _control_disabled(): + return + + global _step_index, _current_step_name + if advance_step: + _step_index += 1 + if step_name: + _current_step_name = step_name + + command = _read_control_command() + if command == "stop": + _write_lifecycle_event("stopping", "任务已停止", step_name=step_name) + raise JobStopped("Job stopped by host.") + if command == "pause": + _write_lifecycle_event("paused", "已暂停，等待继续", step_name=step_name) + _wait_while_paused(step_name=step_name) + + +def checkpoint() -> None: + """ + Consult host control without advancing the step index. + + Use inside long waits (interruptible_sleep) between step boundaries. + """ + _apply_control(advance_step=False) + + +def _step_gate(step_name: str | None) -> None: + """Step boundary: advance index, then pause/stop gate.""" + _apply_control(advance_step=True, step_name=step_name) + + def _build_event( type_: str, text: str, @@ -130,7 +261,7 @@ def _build_event( return event -def emit( +def _emit_core( text: str, *, type: str = "activity", @@ -141,7 +272,6 @@ def emit( item_id: str | None = None, **extra: Any, ) -> None: - """Append one SAS event to the Run Journal. Never writes to stdout.""" if type == "debug" and not _activity_debug_enabled(): return if type not in _VALID_EMIT_TYPES: @@ -159,11 +289,67 @@ def emit( _write_journal_line(event) +def emit( + text: str, + *, + type: str = "activity", + skill: str | None = None, + stage: str | None = None, + current: int | None = None, + total: int | None = None, + item_id: str | None = None, + **extra: Any, +) -> None: + """Append one SAS event to the Run Journal. Never writes to stdout.""" + gate_name = stage or text + _step_gate(gate_name) + _emit_core( + text, + type=type, + skill=skill, + stage=stage, + current=current, + total=total, + item_id=item_id, + **extra, + ) + + def step(text: str, **kwargs: Any) -> None: """Shorthand for emit(type='activity').""" emit(text, type="activity", **kwargs) +async def interruptible_sleep(seconds: float, *, chunk: float | None = None) -> None: + """asyncio.sleep in slices; each slice consults host control (pause/stop).""" + if seconds <= 0: + return + poll = chunk if chunk is not None else _control_poll_sec() + poll = max(0.05, poll) + deadline = time.monotonic() + seconds + while True: + checkpoint() + remaining = deadline - time.monotonic() + if remaining <= 0: + return + await asyncio.sleep(min(poll, remaining)) + + +def interruptible_sleep_sync(seconds: float, *, chunk: float | None = None) -> None: + """time.sleep in slices; each slice consults host control (pause/stop).""" + if seconds <= 0: + return + poll = chunk if chunk is not None else _control_poll_sec() + poll = max(0.05, poll) + deadline = time.monotonic() + seconds + while True: + checkpoint() + remaining = deadline - time.monotonic() + if remaining <= 0: + return + time.sleep(min(poll, remaining)) + + def _serialize_result(result: dict[str, Any]) -> str: line = json.dumps(result, ensure_ascii=False, separators=(",", ":")) if len(line) <= _RESULT_MAX_CHARS: @@ -269,6 +455,14 @@ def job_context( try: yield + except JobStopped as exc: + finish( + status="failed", + message=str(exc) or "用户停止", + skill=skill, + error_code="JOB_STOPPED", + ) + raise except Exception as exc: finish(status="failed", message=str(exc), skill=skill) raise @@ -293,7 +487,7 @@ def job_context( def rpa_step(name: str | None = None) -> Callable[[F], F]: - """Decorator for sync/async RPA steps: emit enter/success/warn on failure.""" + """Decorator for sync/async RPA steps: gate, emit enter/success/warn on failure.""" def decorator(fn: F) -> F: step_name = name or fn.__name__ @@ -302,26 +496,32 @@ def rpa_step(name: str | None = None) -> Callable[[F], F]: @functools.wraps(fn) async def async_wrapper(*args: Any, **kwargs: Any) -> Any: - emit(f"▶ {step_name}") + _step_gate(step_name) + _emit_core(f"▶ {step_name}") try: result = await fn(*args, **kwargs) - emit(f"✓ {step_name}") + _emit_core(f"✓ {step_name}") return result + except JobStopped: + raise except Exception as exc: - emit(f"✗ {step_name}: {exc}", type="warn") + _emit_core(f"✗ {step_name}: {exc}", type="warn") raise return async_wrapper # type: ignore[return-value] @functools.wraps(fn) def sync_wrapper(*args: Any, **kwargs: Any) -> Any: - emit(f"▶ {step_name}") + _step_gate(step_name) + _emit_core(f"▶ {step_name}") try: result = fn(*args, **kwargs) - emit(f"✓ {step_name}") + _emit_core(f"✓ {step_name}") return result + except JobStopped: + raise except Exception as exc: - emit(f"✗ {step_name}: {exc}", type="warn") + _emit_core(f"✗ {step_name}: {exc}", type="warn") raise return sync_wrapper # type: ignore[return-value] @@ -330,8 +530,9 @@ def rpa_step(name: str | None = None) -> Callable[[F], F]: def _reset_for_tests() -> None: - """Clear module-level journal state (tests only).""" + """Clear module-level journal and control state (tests only).""" global _cached_job_id, _journal_path, _meta_written, _journal_file + global _step_index, _current_step_name if _journal_file is not None: try: _journal_file.close() @@ -341,3 +542,5 @@ def _reset_for_tests() -> None: _journal_path = None _meta_written = False _journal_file = None + _step_index = 0 + _current_step_name = None diff --git a/tests/test_activity_control.py b/tests/test_activity_control.py new file mode 100644 index 0000000..2a50ebb --- /dev/null +++ b/tests/test_activity_control.py @@ -0,0 +1,218 @@ +# -*- coding: utf-8 -*- +"""SRCP v1: control.json, checkpoint, pause/resume/stop, interruptible_sleep.""" +from __future__ import annotations + +import asyncio +import io +import json +import os +import shutil +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from jiangchang_skill_core import activity +from jiangchang_skill_core.activity import JobStopped +from jiangchang_skill_core.runtime_env import get_runs_dir + + +class _ControlEnvIsolation(unittest.TestCase): + _KEYS = ( + "JIANGCHANG_DATA_ROOT", + "JIANGCHANG_USER_ID", + "JIANGCHANG_JOB_ID", + "JIANGCHANG_RUN_MODE", + "JIANGCHANG_CONTROL_DISABLED", + "JIANGCHANG_CONTROL_POLL_SEC", + ) + + def setUp(self) -> None: + self._saved = {k: os.environ.get(k) for k in self._KEYS} + self._tmps: list[str] = [] + for k in self._KEYS: + os.environ.pop(k, None) + activity._reset_for_tests() + + def tearDown(self) -> None: + activity._reset_for_tests() + for k, v in self._saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + for tmp in self._tmps: + shutil.rmtree(tmp, ignore_errors=True) + + def _tmp_data_root(self) -> str: + tmp = tempfile.mkdtemp() + self._tmps.append(tmp) + return tmp + + def _write_control(self, job_id: str, command: str, *, nonce: int = 1) -> None: + path = Path(get_runs_dir()) / f"{job_id}.control.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "schema_version": 1, + "command": command, + "reason": "user", + "ts": 1_700_000_000_000, + "nonce": nonce, + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + + def _journal_events(self, job_id: str) -> list[dict]: + journal = Path(get_runs_dir()) / f"{job_id}.jsonl" + if not journal.is_file(): + return [] + return [ + json.loads(line) + for line in journal.read_text(encoding="utf-8").strip().splitlines() + if line.strip() + ] + + +class TestControlProtocol(_ControlEnvIsolation): + def test_get_control_path(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-path" + activity._reset_for_tests() + self.assertTrue(activity.get_control_path().endswith("ctrl-path.control.json")) + + def test_stop_raises_job_stopped(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-stop" + os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05" + activity._reset_for_tests() + self._write_control("ctrl-stop", "stop") + + with self.assertRaises(JobStopped): + activity.checkpoint() + + events = self._journal_events("ctrl-stop") + lifecycle = [e for e in events if e.get("type") == "lifecycle"] + self.assertTrue(any(e.get("status") == "stopping" for e in lifecycle)) + + def test_emit_stop_before_activity_write(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-emit-stop" + activity._reset_for_tests() + self._write_control("ctrl-emit-stop", "stop") + + with self.assertRaises(JobStopped): + activity.emit("不应写入") + + events = self._journal_events("ctrl-emit-stop") + self.assertFalse(any(e.get("type") == "activity" for e in events)) + self.assertTrue(any(e.get("type") == "lifecycle" and e.get("status") == "stopping" for e in events)) + + def test_pause_resume_via_emit(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-pause" + os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05" + activity._reset_for_tests() + self._write_control("ctrl-pause", "pause") + + import threading + import time as time_mod + + def resume_later() -> None: + time_mod.sleep(0.15) + self._write_control("ctrl-pause", "resume", nonce=2) + + threading.Thread(target=resume_later, daemon=True).start() + activity.emit("步骤一", skill="demo") + + events = self._journal_events("ctrl-pause") + statuses = [e.get("status") for e in events if e.get("type") == "lifecycle"] + self.assertIn("paused", statuses) + self.assertIn("resumed", statuses) + activity_events = [e for e in events if e.get("type") == "activity"] + self.assertEqual(activity_events[-1]["text"], "步骤一") + lifecycle = [e for e in events if e.get("type") == "lifecycle"] + self.assertEqual(lifecycle[0].get("step_index"), 1) + + def test_rpa_step_gate_increments_step_index(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-rpa" + activity._reset_for_tests() + + @activity.rpa_step("打开页面") + def open_page() -> None: + return None + + open_page() + events = self._journal_events("ctrl-rpa") + lifecycle_or_activity = [e for e in events if e.get("type") in ("activity", "lifecycle")] + self.assertGreaterEqual(len(lifecycle_or_activity), 2) + + def test_job_context_finishes_on_job_stopped(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + activity._reset_for_tests() + + with patch("sys.stdout", io.StringIO()): + with self.assertRaises(JobStopped): + with activity.job_context(job_id="ctx-stop", skill="demo-skill"): + self._write_control("ctx-stop", "stop") + activity.checkpoint() + + journal = Path(get_runs_dir()) / "ctx-stop.jsonl" + lines = journal.read_text(encoding="utf-8").strip().splitlines() + last = json.loads(lines[-1]) + self.assertEqual(last["type"], "error") + self.assertEqual(last.get("error_code"), "JOB_STOPPED") + + def test_interruptible_sleep_honors_stop(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-sleep" + os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05" + activity._reset_for_tests() + self._write_control("ctrl-sleep", "stop") + + async def run() -> None: + with self.assertRaises(JobStopped): + await activity.interruptible_sleep(5.0) + + asyncio.run(run()) + + def test_interruptible_sleep_sync_honors_pause(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "ctrl-sleep-sync" + os.environ["JIANGCHANG_CONTROL_POLL_SEC"] = "0.05" + activity._reset_for_tests() + self._write_control("ctrl-sleep-sync", "pause") + + def resume_later() -> None: + import threading + import time + + def _resume() -> None: + time.sleep(0.15) + self._write_control("ctrl-sleep-sync", "resume", nonce=2) + + threading.Thread(target=_resume, daemon=True).start() + + resume_later() + activity.interruptible_sleep_sync(0.4, chunk=0.05) + + events = self._journal_events("ctrl-sleep-sync") + statuses = [e.get("status") for e in events if e.get("type") == "lifecycle"] + self.assertIn("paused", statuses) + self.assertIn("resumed", statuses) + + +if __name__ == "__main__": + unittest.main()