diff --git a/README.md b/README.md index d9fb1fa..a0a1f41 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ pip install jiangchang-platform-kit --index-url https://git.jc2009.com/api/packa ## 3. 包含的模块 -- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析、共享媒体资源（`media_assets`）等，供 Skill CLI / 服务进程使用。 +- **jiangchang_skill_core**：权益 HTTP 客户端、`enforce_entitlement`、统一文件日志、`JIANGCHANG_*` 数据根与技能根路径解析、**Skill Activity Stream / Job 运行时**（`emit` / `finish` / Run Journal）、兄弟技能 CLI 桥（`sibling_bridge`）、共享媒体资源（`media_assets`）等，供 Skill CLI / 服务进程使用。 - **jiangchang_desktop_sdk**：匠厂桌面应用自动化客户端（基于 CDP + Playwright 同步 API）。**本包不声明 Playwright 依赖**；请在运行环境中自行安装 Playwright，以便 `import playwright` 可用。 ## 4. 快速开始 @@ -78,6 +78,71 @@ assert_skill_env_file(ctx, ["API_KEY"]) send_prompt_via_composer(page, "第一行\n第二行") ``` +## Skill Activity Stream / Job 运行时 + +长任务（批量生成、RPA）需要把**逐步进度**推到匠厂宿主 UI，而不能依赖 Agent 轮询或 OpenClaw stdout patch。`jiangchang_skill_core.activity` 提供 **Skill Activity Stream (SAS) v1** 契约： + +| API | 作用 | +|-----|------| +| `emit(text, ...)` | 写 Run Journal（`activity` / `progress` / `warn` / `debug`） | +| `step(text)` | `emit(type='activity')` 别名 | +| `finish(status=..., **fields)` | 写终态事件 + **唯一**向 stdout 输出单行 result JSON | +| `rpa_step(name)` | 装饰器：进入/成功/失败自动 emit | +| `job_context(...)` | 可选包裹 main；异常时自动 `finish(failed)` | + +### Run Journal 路径（宿主 tail 此文件） + +``` +{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl +``` + +- `job_id`：优先 `JIANGCHANG_JOB_ID`，否则进程内 `uuid4().hex`（首次 emit 时生成并缓存） +- `JIANGCHANG_RUN_MODE`：`job` | `cli`（默认 `cli`），首行 meta 事件记录 +- NDJSON，UTF-8，append-only，每行 flush +- `debug` 类型仅当 `JIANGCHANG_ACTIVITY_DEBUG=1` 时写入 + +### 与 Agent / OpenClaw 的分工 + +- **`emit()` / journal**：永不 print 到 stdout，避免 bash tool 结果淹没 Agent context +- **`finish()`**：唯一允许向 stdout 写业务结果的 API；单行 compact JSON，`status` + 摘要字段，默认 ≤4096 字符 +- 技能里手写 `print("OK")`、pretty JSON 应逐步改为 `finish()`；RPA 步骤用 `emit("打开浏览器")` 或 `@rpa_step` + +### GEO batch 示例 + +```python +from jiangchang_skill_core.activity import emit, finish + +for i, item in enumerate(items, 1): + emit(f"生成 {item['title']}", type="progress", current=i, total=len(items), skill="geo-batch") + # ... business logic ... + +finish(status="success", skill="geo-batch", generated=len(items)) +``` + +### RPA 步骤示例 + +```python +from jiangchang_skill_core.activity import emit, rpa_step + +@rpa_step("打开浏览器") +def open_browser(page): + page.goto("https://example.com") + +emit("手动步骤说明", skill="publish-toutiao", stage="rpa") +``` + +`RpaVideoSession.add_step()` 在录屏字幕之外默认同步 `emit(..., stage="rpa.video")`；可通过 `emit_activity=False` 关闭。 + +### 兄弟技能 CLI 桥 + +```python +from jiangchang_skill_core import call_sibling_json, get_sibling_main_path + +accounts = call_sibling_json("account-manager", ["list", "all"]) +``` + +同步 `subprocess.run`，`capture_output=True`，自动带上 `subprocess_env_with_trace()`。 + ## 共享媒体资源 `jiangchang_skill_core.media_assets` 提供共享媒体资源解析能力。默认会从： diff --git a/pyproject.toml b/pyproject.toml index 492a2fa..2ab96ca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.0.17" +version = "1.1.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 52e6161..6449f8d 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -1,3 +1,17 @@ +from .activity import ( + emit, + finish, + get_job_id, + get_run_mode, + job_context, + rpa_step, + step, +) +from .sibling_bridge import ( + call_sibling_json, + call_sibling_json_array, + get_sibling_main_path, +) from .client import EntitlementClient from .config import ensure_env_file, get, get_bool, get_float, get_int from .errors import EntitlementDeniedError, EntitlementError, EntitlementServiceError @@ -5,8 +19,10 @@ from .guard import enforce_entitlement from .models import EntitlementResult from .runtime_env import ( apply_cli_local_defaults, + ensure_runs_dir, find_chrome_executable, get_data_root, + get_runs_dir, get_sibling_skills_root, get_skills_root, get_user_id, @@ -52,6 +68,18 @@ except ImportError: rpa = None # type: ignore[assignment,misc] __all__ = [ + "call_sibling_json", + "call_sibling_json_array", + "emit", + "finish", + "get_job_id", + "get_run_mode", + "get_runs_dir", + "ensure_runs_dir", + "job_context", + "rpa_step", + "step", + "get_sibling_main_path", "EntitlementClient", "EntitlementDeniedError", "EntitlementError", @@ -88,6 +116,7 @@ __all__ = [ "get_data_root", "get_float", "get_int", + "get_runs_dir", "get_sibling_skills_root", "get_skills_root", "get_skill_log_file_path", diff --git a/src/jiangchang_skill_core/activity.py b/src/jiangchang_skill_core/activity.py new file mode 100644 index 0000000..e9ae27a --- /dev/null +++ b/src/jiangchang_skill_core/activity.py @@ -0,0 +1,343 @@ +""" +Skill Activity Stream (SAS) v1 — Run Journal + stdout result contract. + +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. +""" + +from __future__ import annotations + +import asyncio +import functools +import inspect +import json +import os +import sys +import time +import uuid +from contextlib import contextmanager +from typing import Any, Callable, Iterator, Optional, TypeVar + +from .runtime_env import ensure_runs_dir, get_runs_dir + +_SCHEMA_VERSION = 1 +_VALID_EMIT_TYPES = frozenset({"activity", "progress", "warn", "debug"}) +_RESULT_MAX_CHARS = 4096 + +_cached_job_id: Optional[str] = None +_journal_path: Optional[str] = None +_meta_written: bool = False +_journal_file: Optional[Any] = None + +F = TypeVar("F", bound=Callable[..., Any]) + + +def _activity_debug_enabled() -> bool: + v = (os.getenv("JIANGCHANG_ACTIVITY_DEBUG") or "").strip().lower() + return v in ("1", "true", "yes", "on") + + +def get_run_mode() -> str: + mode = (os.getenv("JIANGCHANG_RUN_MODE") or "").strip().lower() + if mode in ("job", "cli"): + return mode + return "cli" + + +def get_job_id() -> str: + global _cached_job_id + env_job = (os.getenv("JIANGCHANG_JOB_ID") or "").strip() + if env_job: + return env_job + if _cached_job_id is None: + _cached_job_id = uuid.uuid4().hex + return _cached_job_id + + +def _get_journal_path() -> str: + global _journal_path + if _journal_path is None: + ensure_runs_dir() + _journal_path = os.path.join(get_runs_dir(), f"{get_job_id()}.jsonl") + return _journal_path + + +def _open_journal() -> Any: + global _journal_file + if _journal_file is None: + path = _get_journal_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + _journal_file = open(path, "a", encoding="utf-8") + return _journal_file + + +def _write_journal_line(event: dict[str, Any]) -> None: + global _meta_written + f = _open_journal() + if not _meta_written: + meta = { + "type": "meta", + "schema_version": _SCHEMA_VERSION, + "job_id": get_job_id(), + "run_mode": get_run_mode(), + "ts": _now_ms(), + } + f.write(json.dumps(meta, ensure_ascii=False, separators=(",", ":")) + "\n") + f.flush() + _meta_written = True + f.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n") + f.flush() + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def _build_event( + type_: str, + text: str, + *, + skill: str | None = None, + stage: str | None = None, + current: int | None = None, + total: int | None = None, + item_id: str | None = None, + ts: int | None = None, + **extra: Any, +) -> dict[str, Any]: + cleaned = (text or "").strip() + if not cleaned: + raise ValueError("text must be non-empty") + event: dict[str, Any] = { + "type": type_, + "text": cleaned, + "schema_version": _SCHEMA_VERSION, + "ts": ts if ts is not None else _now_ms(), + } + if skill is not None: + event["skill"] = skill + if stage is not None: + event["stage"] = stage + if current is not None: + event["current"] = current + if total is not None: + event["total"] = total + if item_id is not None: + event["item_id"] = item_id + for key, value in extra.items(): + if key not in event: + event[key] = value + return 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.""" + if type == "debug" and not _activity_debug_enabled(): + return + if type not in _VALID_EMIT_TYPES: + raise ValueError(f"invalid emit type: {type!r}") + event = _build_event( + type, + text, + skill=skill, + stage=stage, + current=current, + total=total, + item_id=item_id, + **extra, + ) + _write_journal_line(event) + + +def step(text: str, **kwargs: Any) -> None: + """Shorthand for emit(type='activity').""" + emit(text, type="activity", **kwargs) + + +def _serialize_result(result: dict[str, Any]) -> str: + line = json.dumps(result, ensure_ascii=False, separators=(",", ":")) + if len(line) <= _RESULT_MAX_CHARS: + return line + + core_keys = ("type", "schema_version", "status", "skill", "message") + truncated: dict[str, Any] = { + k: result[k] for k in core_keys if k in result + } + truncated["truncated"] = True + + for key, value in result.items(): + if key in truncated or key == "truncated": + continue + candidate = {**truncated, key: value} + candidate_line = json.dumps(candidate, ensure_ascii=False, separators=(",", ":")) + if len(candidate_line) <= _RESULT_MAX_CHARS: + truncated[key] = value + + line = json.dumps(truncated, ensure_ascii=False, separators=(",", ":")) + if len(line) > _RESULT_MAX_CHARS: + truncated.pop("message", None) + line = json.dumps(truncated, ensure_ascii=False, separators=(",", ":")) + if len(line) > _RESULT_MAX_CHARS: + return line[:_RESULT_MAX_CHARS] + return line + + +def finish( + *, + status: str = "success", + message: str | None = None, + skill: str | None = None, + **fields: Any, +) -> None: + """ + 1) Write terminal done|error event to Run Journal. + 2) Print a single compact result JSON line to stdout (Agent/bash path). + """ + if status not in ("success", "partial", "failed"): + raise ValueError(f"invalid finish status: {status!r}") + + journal_type = "error" if status == "failed" else "done" + terminal_text = message or ("任务失败" if status == "failed" else "任务完成") + + journal_event = _build_event( + journal_type, + terminal_text, + skill=skill, + status=status, + **fields, + ) + if message is not None: + journal_event["message"] = message + _write_journal_line(journal_event) + + result: dict[str, Any] = { + "type": "result", + "schema_version": _SCHEMA_VERSION, + "status": status, + } + if skill is not None: + result["skill"] = skill + if message is not None: + result["message"] = message + result.update(fields) + + line = _serialize_result(result) + sys.stdout.write(line + "\n") + sys.stdout.flush() + + +@contextmanager +def job_context( + *, + job_id: str | None = None, + skill: str | None = None, + run_mode: str | None = None, +) -> Iterator[None]: + """Optional wrapper: sets job context; auto finish(failed) on unhandled exception.""" + global _cached_job_id, _journal_path, _meta_written, _journal_file + + saved_job_id = _cached_job_id + saved_journal_path = _journal_path + saved_meta = _meta_written + saved_file = _journal_file + saved_env_job = os.environ.get("JIANGCHANG_JOB_ID") + saved_env_mode = os.environ.get("JIANGCHANG_RUN_MODE") + + if job_id: + os.environ["JIANGCHANG_JOB_ID"] = job_id + _cached_job_id = None + _journal_path = None + _meta_written = False + if _journal_file is not None: + try: + _journal_file.close() + except Exception: + pass + _journal_file = None + if run_mode: + os.environ["JIANGCHANG_RUN_MODE"] = run_mode + + try: + yield + except Exception as exc: + finish(status="failed", message=str(exc), skill=skill) + raise + finally: + _cached_job_id = saved_job_id + _journal_path = saved_journal_path + _meta_written = saved_meta + if _journal_file is not None and _journal_file is not saved_file: + try: + _journal_file.close() + except Exception: + pass + _journal_file = saved_file + if saved_env_job is None: + os.environ.pop("JIANGCHANG_JOB_ID", None) + else: + os.environ["JIANGCHANG_JOB_ID"] = saved_env_job + if saved_env_mode is None: + os.environ.pop("JIANGCHANG_RUN_MODE", None) + else: + os.environ["JIANGCHANG_RUN_MODE"] = saved_env_mode + + +def rpa_step(name: str | None = None) -> Callable[[F], F]: + """Decorator for sync/async RPA steps: emit enter/success/warn on failure.""" + + def decorator(fn: F) -> F: + step_name = name or fn.__name__ + + if inspect.iscoroutinefunction(fn): + + @functools.wraps(fn) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + emit(f"▶ {step_name}") + try: + result = await fn(*args, **kwargs) + emit(f"✓ {step_name}") + return result + except Exception as exc: + emit(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}") + try: + result = fn(*args, **kwargs) + emit(f"✓ {step_name}") + return result + except Exception as exc: + emit(f"✗ {step_name}: {exc}", type="warn") + raise + + return sync_wrapper # type: ignore[return-value] + + return decorator + + +def _reset_for_tests() -> None: + """Clear module-level journal state (tests only).""" + global _cached_job_id, _journal_path, _meta_written, _journal_file + if _journal_file is not None: + try: + _journal_file.close() + except Exception: + pass + _cached_job_id = None + _journal_path = None + _meta_written = False + _journal_file = None diff --git a/src/jiangchang_skill_core/rpa/video_session.py b/src/jiangchang_skill_core/rpa/video_session.py index 5cb44d9..fb3c6c6 100644 --- a/src/jiangchang_skill_core/rpa/video_session.py +++ b/src/jiangchang_skill_core/rpa/video_session.py @@ -568,12 +568,14 @@ class RpaVideoSession: batch_id: str, title: str = "", closing_title: str = "", + emit_activity: bool = True, ) -> None: self.skill_slug = skill_slug self.skill_data_dir = os.path.abspath(skill_data_dir) self.batch_id = batch_id self.title = title self.closing_title = closing_title + self.emit_activity = emit_activity self.enabled = _record_video_enabled() self.warnings: List[str] = [] self.output_video_path: Optional[str] = None @@ -631,13 +633,25 @@ class RpaVideoSession: self.add_step(text, duration=duration) def add_step(self, text: str, *, duration: float = 4.0) -> None: - """记录一步骤字幕（相对录屏开始时间）。""" + """记录一步骤字幕（相对录屏开始时间）；可选同步写入 Activity Stream。""" + cleaned = text.strip() + if self.emit_activity and cleaned: + try: + from .. import activity + + activity.emit( + cleaned, + skill=self.skill_slug, + stage="rpa.video", + ) + except Exception: + pass if not self.enabled: return if self._t0 is None: self._t0 = time.monotonic() elapsed = time.monotonic() - self._t0 - self._steps.append(_StepEntry(start=elapsed, text=text.strip(), duration=duration)) + self._steps.append(_StepEntry(start=elapsed, text=cleaned, duration=duration)) def _close_ffmpeg_record_log(self) -> None: fh = self._ffmpeg_record_log_file diff --git a/src/jiangchang_skill_core/runtime_env.py b/src/jiangchang_skill_core/runtime_env.py index 1f68fa2..3b2b4e2 100644 --- a/src/jiangchang_skill_core/runtime_env.py +++ b/src/jiangchang_skill_core/runtime_env.py @@ -44,6 +44,20 @@ def get_user_id() -> str: return (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon" +def get_runs_dir() -> str: + """Run Journal 目录：{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/""" + return os.path.join(get_data_root(), ".jiangchang", "runs") + + +def ensure_runs_dir() -> str: + path = get_runs_dir() + try: + os.makedirs(path, exist_ok=True) + except OSError: + pass + return path + + def _looks_like_skills_root(path: str) -> bool: if not path or not os.path.isdir(path): return False diff --git a/src/jiangchang_skill_core/sibling_bridge.py b/src/jiangchang_skill_core/sibling_bridge.py new file mode 100644 index 0000000..0481fd8 --- /dev/null +++ b/src/jiangchang_skill_core/sibling_bridge.py @@ -0,0 +1,71 @@ +""" +兄弟技能 CLI 桥：同步 subprocess 调用并列技能 scripts/main.py。 + +account-manager 等兄弟技能为同步 CLI，非 async、非线程池。 +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from typing import Any, List, Optional, Sequence + +from .runtime_env import get_skills_root +from .unified_logging import subprocess_env_with_trace + + +def get_sibling_main_path(skill_slug: str) -> str: + return os.path.join(get_skills_root(), skill_slug, "scripts", "main.py") + + +def _python_exe() -> str: + return sys.executable + + +def _run_sibling(skill_slug: str, argv: Sequence[str]) -> subprocess.CompletedProcess[str]: + main_path = get_sibling_main_path(skill_slug) + cmd = [_python_exe(), main_path, *argv] + return subprocess.run( + cmd, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + env=subprocess_env_with_trace(), + ) + + +def _parse_json_from_stdout(stdout: str) -> Any: + for line in reversed((stdout or "").splitlines()): + stripped = line.strip() + if not stripped or stripped.startswith("ERROR:"): + continue + try: + return json.loads(stripped) + except json.JSONDecodeError: + continue + return None + + +def call_sibling_json(skill_slug: str, argv: Sequence[str]) -> dict | None: + """Run sibling CLI; return last JSON object line from stdout, or None.""" + result = _run_sibling(skill_slug, argv) + if result.returncode != 0: + return None + parsed = _parse_json_from_stdout(result.stdout or "") + if isinstance(parsed, dict): + return parsed + return None + + +def call_sibling_json_array(skill_slug: str, argv: Sequence[str]) -> list | None: + """Run sibling CLI; return last JSON array line from stdout, or None.""" + result = _run_sibling(skill_slug, argv) + if result.returncode != 0: + return None + parsed = _parse_json_from_stdout(result.stdout or "") + if isinstance(parsed, list): + return parsed + return None diff --git a/tests/test_activity_emit_finish.py b/tests/test_activity_emit_finish.py new file mode 100644 index 0000000..7e8a89b --- /dev/null +++ b/tests/test_activity_emit_finish.py @@ -0,0 +1,209 @@ +# -*- coding: utf-8 -*- +"""Skill Activity Stream: journal, flush, job_id, finish stdout, truncation.""" +from __future__ import annotations + +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.runtime_env import get_runs_dir + + +class _ActivityEnvIsolation(unittest.TestCase): + _KEYS = ( + "JIANGCHANG_DATA_ROOT", + "JIANGCHANG_USER_ID", + "JIANGCHANG_JOB_ID", + "JIANGCHANG_RUN_MODE", + "JIANGCHANG_ACTIVITY_DEBUG", + ) + + 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 + + +class TestActivityJournal(_ActivityEnvIsolation): + def test_emit_writes_ndjson_with_meta(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-test-1" + activity._reset_for_tests() + + activity.emit("步骤一", type="activity", skill="geo-batch") + journal = Path(get_runs_dir()) / "job-test-1.jsonl" + self.assertTrue(journal.is_file()) + lines = journal.read_text(encoding="utf-8").strip().splitlines() + self.assertGreaterEqual(len(lines), 2) + meta = json.loads(lines[0]) + self.assertEqual(meta["type"], "meta") + self.assertEqual(meta["job_id"], "job-test-1") + event = json.loads(lines[1]) + self.assertEqual(event["type"], "activity") + self.assertEqual(event["text"], "步骤一") + self.assertEqual(event["schema_version"], 1) + self.assertEqual(event["skill"], "geo-batch") + self.assertIn("ts", event) + + def test_emit_flushes_immediately(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-flush" + activity._reset_for_tests() + activity.emit("flush test") + journal = Path(get_runs_dir()) / "job-flush.jsonl" + self.assertGreater(journal.stat().st_size, 0) + + def test_job_id_from_env(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "env-job-42" + activity._reset_for_tests() + self.assertEqual(activity.get_job_id(), "env-job-42") + + def test_job_id_generated_and_cached(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + activity._reset_for_tests() + jid1 = activity.get_job_id() + jid2 = activity.get_job_id() + self.assertEqual(jid1, jid2) + self.assertEqual(len(jid1), 32) + + def test_debug_skipped_without_env(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-no-debug" + activity._reset_for_tests() + activity.emit("hidden", type="debug") + journal = Path(get_runs_dir()) / "job-no-debug.jsonl" + self.assertFalse(journal.exists()) + + def test_debug_written_when_enabled(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-debug" + os.environ["JIANGCHANG_ACTIVITY_DEBUG"] = "1" + activity._reset_for_tests() + activity.emit("dbg", type="debug") + journal = Path(get_runs_dir()) / "job-debug.jsonl" + content = journal.read_text(encoding="utf-8") + self.assertIn('"type":"debug"', content.replace(" ", "")) + + def test_emit_never_prints_stdout(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-quiet" + activity._reset_for_tests() + buf = io.StringIO() + with patch("sys.stdout", buf): + activity.emit("silent step") + self.assertEqual(buf.getvalue(), "") + + def test_get_run_mode_default_cli(self) -> None: + self.assertEqual(activity.get_run_mode(), "cli") + + def test_get_run_mode_job(self) -> None: + os.environ["JIANGCHANG_RUN_MODE"] = "job" + self.assertEqual(activity.get_run_mode(), "job") + + +class TestActivityFinish(_ActivityEnvIsolation): + def test_finish_stdout_single_line_result(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-finish" + activity._reset_for_tests() + + buf = io.StringIO() + with patch("sys.stdout", buf): + activity.finish(status="success", message="ok", skill="geo", generated=3) + + lines = [ln for ln in buf.getvalue().splitlines() if ln.strip()] + self.assertEqual(len(lines), 1) + result = json.loads(lines[0]) + self.assertEqual(result["type"], "result") + self.assertEqual(result["schema_version"], 1) + self.assertEqual(result["status"], "success") + self.assertEqual(result["skill"], "geo") + self.assertEqual(result["generated"], 3) + + journal = Path(get_runs_dir()) / "job-finish.jsonl" + journal_lines = journal.read_text(encoding="utf-8").strip().splitlines() + last = json.loads(journal_lines[-1]) + self.assertEqual(last["type"], "done") + self.assertEqual(last["status"], "success") + + def test_finish_failed_writes_error_event(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-fail" + activity._reset_for_tests() + + with patch("sys.stdout", io.StringIO()): + activity.finish(status="failed", message="boom") + + journal = Path(get_runs_dir()) / "job-fail.jsonl" + last = json.loads(journal.read_text(encoding="utf-8").strip().splitlines()[-1]) + self.assertEqual(last["type"], "error") + + def test_finish_truncates_large_result(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "job-trunc" + activity._reset_for_tests() + + big = {f"field_{i}": "x" * 200 for i in range(50)} + buf = io.StringIO() + with patch("sys.stdout", buf): + activity.finish(status="success", skill="big", **big) + + line = buf.getvalue().strip() + self.assertLessEqual(len(line), 4096) + result = json.loads(line) + self.assertTrue(result.get("truncated")) + + +class TestJobContext(_ActivityEnvIsolation): + def test_job_context_auto_finish_on_exception(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(ValueError): + with activity.job_context(job_id="ctx-err", skill="test-skill"): + raise ValueError("broken") + + journal = Path(get_runs_dir()) / "ctx-err.jsonl" + last = json.loads(journal.read_text(encoding="utf-8").strip().splitlines()[-1]) + self.assertEqual(last["type"], "error") + self.assertIn("broken", last["message"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_activity_rpa_step_decorator.py b/tests/test_activity_rpa_step_decorator.py new file mode 100644 index 0000000..763a793 --- /dev/null +++ b/tests/test_activity_rpa_step_decorator.py @@ -0,0 +1,120 @@ +# -*- coding: utf-8 -*- +"""rpa_step decorator: sync + async.""" +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import tempfile +import unittest +from pathlib import Path + +from jiangchang_skill_core import activity +from jiangchang_skill_core.runtime_env import get_runs_dir + + +class _DecoratorEnvIsolation(unittest.TestCase): + def setUp(self) -> None: + self._saved_root = os.environ.get("JIANGCHANG_DATA_ROOT") + self._saved_job = os.environ.get("JIANGCHANG_JOB_ID") + self._tmps: list[str] = [] + os.environ.pop("JIANGCHANG_JOB_ID", None) + activity._reset_for_tests() + + def tearDown(self) -> None: + activity._reset_for_tests() + if self._saved_root is None: + os.environ.pop("JIANGCHANG_DATA_ROOT", None) + else: + os.environ["JIANGCHANG_DATA_ROOT"] = self._saved_root + if self._saved_job is None: + os.environ.pop("JIANGCHANG_JOB_ID", None) + else: + os.environ["JIANGCHANG_JOB_ID"] = self._saved_job + 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 + + +class TestRpaStepDecorator(_DecoratorEnvIsolation): + def _journal_texts(self, job_id: str) -> list[str]: + journal = Path(get_runs_dir()) / f"{job_id}.jsonl" + lines = journal.read_text(encoding="utf-8").strip().splitlines() + events = [json.loads(ln) for ln in lines if json.loads(ln).get("type") != "meta"] + return [e["text"] for e in events] + + def test_sync_success(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "rpa-sync-ok" + activity._reset_for_tests() + + @activity.rpa_step("打开浏览器") + def open_browser() -> str: + return "ok" + + self.assertEqual(open_browser(), "ok") + texts = self._journal_texts("rpa-sync-ok") + self.assertIn("▶ 打开浏览器", texts) + self.assertIn("✓ 打开浏览器", texts) + + def test_sync_failure_emits_warn_and_reraises(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "rpa-sync-fail" + activity._reset_for_tests() + + @activity.rpa_step("提交表单") + def submit() -> None: + raise RuntimeError("timeout") + + with self.assertRaises(RuntimeError): + submit() + + journal = Path(get_runs_dir()) / "rpa-sync-fail.jsonl" + content = journal.read_text(encoding="utf-8") + self.assertIn("timeout", content) + self.assertIn('"type":"warn"', content.replace(" ", "")) + + def test_async_success(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "rpa-async-ok" + activity._reset_for_tests() + + @activity.rpa_step("异步步骤") + async def async_step() -> int: + await asyncio.sleep(0) + return 42 + + result = asyncio.run(async_step()) + self.assertEqual(result, 42) + texts = self._journal_texts("rpa-async-ok") + self.assertIn("▶ 异步步骤", texts) + self.assertIn("✓ 异步步骤", texts) + + def test_async_failure(self) -> None: + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "rpa-async-fail" + activity._reset_for_tests() + + @activity.rpa_step() + async def fail_step() -> None: + await asyncio.sleep(0) + raise ValueError("async err") + + with self.assertRaises(ValueError): + asyncio.run(fail_step()) + + journal = Path(get_runs_dir()) / "rpa-async-fail.jsonl" + self.assertIn("async err", journal.read_text(encoding="utf-8")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sibling_bridge.py b/tests/test_sibling_bridge.py new file mode 100644 index 0000000..f64233f --- /dev/null +++ b/tests/test_sibling_bridge.py @@ -0,0 +1,72 @@ +# -*- coding: utf-8 -*- +"""sibling_bridge: mock subprocess sibling CLI calls.""" +from __future__ import annotations + +import json +import os +import subprocess +import unittest +from unittest.mock import MagicMock, patch + +from jiangchang_skill_core.sibling_bridge import ( + call_sibling_json, + call_sibling_json_array, + get_sibling_main_path, +) + + +class TestSiblingBridge(unittest.TestCase): + def test_get_sibling_main_path(self) -> None: + with patch.dict(os.environ, {"JIANGCHANG_SKILLS_ROOT": r"D:\skills"}): + path = get_sibling_main_path("account-manager") + self.assertTrue(path.replace("\\", "/").endswith("account-manager/scripts/main.py")) + + @patch("jiangchang_skill_core.sibling_bridge.subprocess.run") + def test_call_sibling_json_success(self, mock_run: MagicMock) -> None: + mock_run.return_value = subprocess.CompletedProcess( + args=[], + returncode=0, + stdout='log line\n{"id": 1, "name": "test"}\n', + stderr="", + ) + result = call_sibling_json("account-manager", ["list", "all"]) + self.assertEqual(result, {"id": 1, "name": "test"}) + cmd = mock_run.call_args[0][0] + self.assertIn("account-manager", cmd[1]) + self.assertIn("list", cmd) + self.assertTrue(mock_run.call_args[1]["capture_output"]) + + @patch("jiangchang_skill_core.sibling_bridge.subprocess.run") + def test_call_sibling_json_nonzero_exit(self, mock_run: MagicMock) -> None: + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr="ERROR:fail" + ) + self.assertIsNone(call_sibling_json("content-manager", ["get", "x"])) + + @patch("jiangchang_skill_core.sibling_bridge.subprocess.run") + def test_call_sibling_json_array(self, mock_run: MagicMock) -> None: + mock_run.return_value = subprocess.CompletedProcess( + args=[], + returncode=0, + stdout=json.dumps([{"a": 1}, {"b": 2}]), + stderr="", + ) + result = call_sibling_json_array("account-manager", ["list", "all"]) + self.assertEqual(result, [{"a": 1}, {"b": 2}]) + + @patch("jiangchang_skill_core.sibling_bridge.subprocess.run") + def test_uses_subprocess_env_with_trace(self, mock_run: MagicMock) -> None: + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=0, stdout='{"ok":true}', stderr="" + ) + with patch( + "jiangchang_skill_core.sibling_bridge.subprocess_env_with_trace", + return_value={"JIANGCHANG_TRACE_ID": "abc123"}, + ) as mock_env: + call_sibling_json("api-key-vault", ["get", "key"]) + self.assertEqual(mock_run.call_args[1]["env"], {"JIANGCHANG_TRACE_ID": "abc123"}) + mock_env.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_video_session_activity_integration.py b/tests/test_video_session_activity_integration.py new file mode 100644 index 0000000..9389734 --- /dev/null +++ b/tests/test_video_session_activity_integration.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*- +"""RpaVideoSession.add_step integrates with activity.emit.""" +from __future__ import annotations + +import json +import os +import shutil +import tempfile +import unittest +from pathlib import Path + +from jiangchang_skill_core import activity, config +from jiangchang_skill_core.rpa.video_session import RpaVideoSession +from jiangchang_skill_core.runtime_env import get_runs_dir + + +class TestVideoSessionActivityIntegration(unittest.TestCase): + def setUp(self) -> None: + config.reset_cache() + self._saved_root = os.environ.get("JIANGCHANG_DATA_ROOT") + self._saved_job = os.environ.get("JIANGCHANG_JOB_ID") + self._saved_video = os.environ.get("OPENCLAW_RECORD_VIDEO") + self._tmps: list[str] = [] + activity._reset_for_tests() + + def tearDown(self) -> None: + activity._reset_for_tests() + if self._saved_root is None: + os.environ.pop("JIANGCHANG_DATA_ROOT", None) + else: + os.environ["JIANGCHANG_DATA_ROOT"] = self._saved_root + if self._saved_job is None: + os.environ.pop("JIANGCHANG_JOB_ID", None) + else: + os.environ["JIANGCHANG_JOB_ID"] = self._saved_job + if self._saved_video is None: + os.environ.pop("OPENCLAW_RECORD_VIDEO", None) + else: + os.environ["OPENCLAW_RECORD_VIDEO"] = self._saved_video + config.reset_cache() + 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 test_add_step_emits_activity_when_disabled(self) -> None: + os.environ.pop("OPENCLAW_RECORD_VIDEO", None) + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "video-act-1" + activity._reset_for_tests() + config.reset_cache() + + session = RpaVideoSession( + skill_slug="publish-toutiao", + skill_data_dir=tmp, + batch_id="batch1", + ) + self.assertFalse(session.enabled) + session.add_step("打开浏览器") + + journal = Path(get_runs_dir()) / "video-act-1.jsonl" + self.assertTrue(journal.is_file()) + content = journal.read_text(encoding="utf-8") + self.assertIn("打开浏览器", content) + self.assertIn("rpa.video", content) + self.assertIn("publish-toutiao", content) + + def test_emit_activity_false_skips_journal(self) -> None: + os.environ.pop("OPENCLAW_RECORD_VIDEO", None) + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "video-act-off" + activity._reset_for_tests() + + session = RpaVideoSession( + skill_slug="publish-toutiao", + skill_data_dir=tmp, + batch_id="batch2", + emit_activity=False, + ) + session.add_step("不应写入") + journal = Path(get_runs_dir()) / "video-act-off.jsonl" + self.assertFalse(journal.exists()) + + def test_add_step_writes_both_journal_and_subtitle_steps(self) -> None: + os.environ["OPENCLAW_RECORD_VIDEO"] = "1" + tmp = self._tmp_data_root() + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_JOB_ID"] = "video-act-both" + activity._reset_for_tests() + config.reset_cache() + + session = RpaVideoSession( + skill_slug="receive-order", + skill_data_dir=tmp, + batch_id="batch3", + ) + session.add_step("录制步骤") + self.assertEqual(len(session._steps), 1) + + journal = Path(get_runs_dir()) / "video-act-both.jsonl" + lines = [json.loads(ln) for ln in journal.read_text(encoding="utf-8").splitlines()] + activity_events = [e for e in lines if e.get("type") == "activity"] + self.assertEqual(len(activity_events), 1) + self.assertEqual(activity_events[0]["text"], "录制步骤") + + +if __name__ == "__main__": + unittest.main()