3 Commits

Author SHA1 Message Date
3dc46b0555 fix(config): treat empty .env values with inline # comments as empty
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 32s
2026-07-18 15:53:37 +08:00
54b75c641e feat(release): sync skill detail tabs and publish 1.2.1
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 26s
Package optional TUTORIAL/DEMO/CHANGELOG into skill ZIPs and sync tutorial_md, demo_md, demo_video_url, and versioned changelog to the marketplace API.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-17 11:48:02 +08:00
cf5e74d80a feat(activity): SRCP v1 step gates, pause/resume/stop control (v1.2.0)
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 20s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-12 12:20:37 +08:00
9 changed files with 733 additions and 58 deletions

View File

@@ -115,6 +115,12 @@ jobs:
" print('Copied README.md')" \ " print('Copied README.md')" \
'if os.path.isfile(readme_src) and not os.path.isfile(readme_dst):' \ 'if os.path.isfile(readme_src) and not os.path.isfile(readme_dst):' \
" raise RuntimeError('README.md exists in skill root but was not packaged')" \ " raise RuntimeError('README.md exists in skill root but was not packaged')" \
'for market_doc in ("TUTORIAL.md", "DEMO.md", "CHANGELOG.md"):' \
' src = os.path.abspath(market_doc)' \
' dst = os.path.join(PACKAGE, market_doc)' \
' if os.path.isfile(src):' \
' shutil.copy2(src, dst)' \
" print(f'Copied {market_doc}')" \
"copy_dir('references', os.path.join(PACKAGE, 'references'), references_ignore)" \ "copy_dir('references', os.path.join(PACKAGE, 'references'), references_ignore)" \
"copy_dir('assets', os.path.join(PACKAGE, 'assets'), common_ignore)" \ "copy_dir('assets', os.path.join(PACKAGE, 'assets'), common_ignore)" \
"copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \ "copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \
@@ -144,49 +150,119 @@ jobs:
- name: Parse Metadata and Pack - name: Parse Metadata and Pack
id: build_task id: build_task
run: | run: |
python3.12 -c " cat > /tmp/parse_skill_release_metadata.py <<'PY'
import frontmatter, os, json, shutil import frontmatter
post = frontmatter.load('SKILL.md') import json
import os
import re
import shutil
def load_md(path: str):
post = frontmatter.load(path)
meta = post.metadata or {}
body = (post.content or "").strip()
return meta if isinstance(meta, dict) else {}, body
def extract_changelog_for_version(text: str, version: str) -> str | None:
"""Extract Keep-a-Changelog / plain ## version section; None if not found."""
if not text or not version:
return None
ver = re.escape(version.lstrip("vV"))
pattern = re.compile(
rf"^##\s*\[?v?{ver}\]?([^\n]*)$",
re.MULTILINE | re.IGNORECASE,
)
match = pattern.search(text)
if not match:
return None
start = match.end()
next_h2 = re.search(r"^##\s+", text[start:], re.MULTILINE)
end = start + next_h2.start() if next_h2 else len(text)
return text[start:end].strip()
post = frontmatter.load("SKILL.md")
metadata = dict(post.metadata or {}) metadata = dict(post.metadata or {})
skill_readme_md = (post.content or '').strip() skill_readme_md = (post.content or "").strip()
skill_description = metadata.get('description') skill_description = metadata.get("description")
metadata['readme_md'] = skill_readme_md metadata["readme_md"] = skill_readme_md
readme_path = 'README.md'
if os.path.isfile(readme_path): if os.path.isfile("README.md"):
try: try:
readme_post = frontmatter.load(readme_path) rm_meta, body = load_md("README.md")
body = (readme_post.content or '').strip() desc = rm_meta.get("description")
rm_meta = readme_post.metadata or {}
desc = rm_meta.get('description')
if desc is not None and str(desc).strip(): if desc is not None and str(desc).strip():
metadata['description'] = str(desc).strip() metadata["description"] = str(desc).strip()
if body: if body:
metadata['readme_md'] = body metadata["readme_md"] = body
except Exception: except Exception:
pass pass
if not (metadata.get('readme_md') or '').strip():
metadata['readme_md'] = skill_readme_md if not (metadata.get("readme_md") or "").strip():
if skill_description is not None and not str(metadata.get('description', '') or '').strip(): metadata["readme_md"] = skill_readme_md
metadata['description'] = skill_description if skill_description is not None and not str(metadata.get("description", "") or "").strip():
openclaw_meta = metadata.get('metadata', {}).get('openclaw', {}) metadata["description"] = skill_description
slug = (openclaw_meta.get('slug') or metadata.get('slug') or metadata.get('name') or '').strip()
# Market tabs: only set keys when files exist (backend updates only present keys).
if os.path.isfile("TUTORIAL.md"):
try:
_tm_meta, tutorial_body = load_md("TUTORIAL.md")
metadata["tutorial_md"] = tutorial_body
except Exception as exc:
raise RuntimeError(f"Failed to parse TUTORIAL.md: {exc}") from exc
if os.path.isfile("DEMO.md"):
try:
demo_meta, demo_body = load_md("DEMO.md")
metadata["demo_md"] = demo_body
url = demo_meta.get("demo_video_url")
metadata["demo_video_url"] = (
str(url).strip() if url is not None and str(url).strip() else ""
)
except Exception as exc:
raise RuntimeError(f"Failed to parse DEMO.md: {exc}") from exc
openclaw_meta = metadata.get("metadata", {}).get("openclaw", {})
if not isinstance(openclaw_meta, dict):
openclaw_meta = {}
slug = (
openclaw_meta.get("slug")
or metadata.get("slug")
or metadata.get("name")
or ""
).strip()
if not slug: if not slug:
raise Exception('SKILL.md 缺少 slug/name') raise Exception("SKILL.md 缺少 slug/name")
ref_name = (os.environ.get('GITHUB_REF_NAME') or '').strip()
if not ref_name.startswith('v'): ref_name = (os.environ.get("GITHUB_REF_NAME") or "").strip()
raise Exception(f'非法标签: {ref_name}') if not ref_name.startswith("v"):
version = ref_name.lstrip('v') raise Exception(f"非法标签: {ref_name}")
metadata['version'] = version version = ref_name.lstrip("v")
artifact_platform = (os.environ.get('ARTIFACT_PLATFORM') or 'windows').strip() metadata["version"] = version
zip_base = f'{slug}-{artifact_platform}'
with open(os.environ['GITHUB_OUTPUT'], 'a') as f: if os.path.isfile("CHANGELOG.md"):
f.write(f'slug={slug}\n') try:
f.write(f'version={version}\n') with open("CHANGELOG.md", encoding="utf-8") as fh:
f.write(f'zip_base={zip_base}\n') changelog_text = fh.read()
f.write(f'artifact_platform={artifact_platform}\n') section = extract_changelog_for_version(changelog_text, version)
f.write(f'metadata={json.dumps(metadata, ensure_ascii=False)}\n') if section is not None:
shutil.make_archive(zip_base, 'zip', 'dist/package') metadata["changelog"] = section
" except Exception as exc:
raise RuntimeError(f"Failed to parse CHANGELOG.md: {exc}") from exc
artifact_platform = (os.environ.get("ARTIFACT_PLATFORM") or "windows").strip()
zip_base = f"{slug}-{artifact_platform}"
with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as f:
f.write(f"slug={slug}\n")
f.write(f"version={version}\n")
f.write(f"zip_base={zip_base}\n")
f.write(f"artifact_platform={artifact_platform}\n")
f.write(f"metadata={json.dumps(metadata, ensure_ascii=False)}\n")
shutil.make_archive(zip_base, "zip", "dist/package")
PY
python3.12 /tmp/parse_skill_release_metadata.py
- name: Sync Database - name: Sync Database
env: env:

View File

@@ -87,7 +87,10 @@ send_prompt_via_composer(page, "第一行\n第二行")
| `emit(text, ...)` | 写 Run Journal`activity` / `progress` / `warn` / `debug` | | `emit(text, ...)` | 写 Run Journal`activity` / `progress` / `warn` / `debug` |
| `step(text)` | `emit(type='activity')` 别名 | | `step(text)` | `emit(type='activity')` 别名 |
| `finish(status=..., **fields)` | 写终态事件 + **唯一**向 stdout 输出单行 result JSON | | `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)` | | `job_context(...)` | 可选包裹 main异常时自动 `finish(failed)` |
### Run Journal 路径(宿主 tail 此文件) ### Run Journal 路径(宿主 tail 此文件)
@@ -133,6 +136,95 @@ emit("手动步骤说明", skill="publish-toutiao", stage="rpa")
`RpaVideoSession.add_step()` 在录屏字幕之外默认同步 `emit(..., stage="rpa.video")`;可通过 `emit_activity=False` 关闭。 `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 桥 ### 兄弟技能 CLI 桥
```python ```python

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "jiangchang-platform-kit" name = "jiangchang-platform-kit"
version = "1.1.0" version = "1.2.2"
description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK"
readme = "README.md" readme = "README.md"
requires-python = ">=3.10" requires-python = ">=3.10"

View File

@@ -1,8 +1,13 @@
from .activity import ( from .activity import (
JobStopped,
checkpoint,
emit, emit,
finish, finish,
get_control_path,
get_job_id, get_job_id,
get_run_mode, get_run_mode,
interruptible_sleep,
interruptible_sleep_sync,
job_context, job_context,
rpa_step, rpa_step,
step, step,
@@ -68,14 +73,19 @@ except ImportError:
rpa = None # type: ignore[assignment,misc] rpa = None # type: ignore[assignment,misc]
__all__ = [ __all__ = [
"JobStopped",
"call_sibling_json", "call_sibling_json",
"call_sibling_json_array", "call_sibling_json_array",
"checkpoint",
"emit", "emit",
"finish", "finish",
"get_control_path",
"get_job_id", "get_job_id",
"get_run_mode", "get_run_mode",
"get_runs_dir", "get_runs_dir",
"ensure_runs_dir", "ensure_runs_dir",
"interruptible_sleep",
"interruptible_sleep_sync",
"job_context", "job_context",
"rpa_step", "rpa_step",
"step", "step",

View File

@@ -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). 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. 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 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 from .runtime_env import ensure_runs_dir, get_runs_dir
_SCHEMA_VERSION = 1 _SCHEMA_VERSION = 1
_CONTROL_SCHEMA_VERSION = 1
_VALID_EMIT_TYPES = frozenset({"activity", "progress", "warn", "debug"}) _VALID_EMIT_TYPES = frozenset({"activity", "progress", "warn", "debug"})
_VALID_CONTROL_COMMANDS = frozenset({"none", "pause", "resume", "stop"})
_RESULT_MAX_CHARS = 4096 _RESULT_MAX_CHARS = 4096
_DEFAULT_CONTROL_POLL_SEC = 0.2
_cached_job_id: Optional[str] = None _cached_job_id: Optional[str] = None
_journal_path: Optional[str] = None _journal_path: Optional[str] = None
_meta_written: bool = False _meta_written: bool = False
_journal_file: Optional[Any] = None _journal_file: Optional[Any] = None
_step_index: int = 0
_current_step_name: Optional[str] = None
F = TypeVar("F", bound=Callable[..., Any]) F = TypeVar("F", bound=Callable[..., Any])
class JobStopped(Exception):
"""Raised when the host requests stop via control.json."""
def _activity_debug_enabled() -> bool: def _activity_debug_enabled() -> bool:
v = (os.getenv("JIANGCHANG_ACTIVITY_DEBUG") or "").strip().lower() v = (os.getenv("JIANGCHANG_ACTIVITY_DEBUG") or "").strip().lower()
return v in ("1", "true", "yes", "on") 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: def get_run_mode() -> str:
mode = (os.getenv("JIANGCHANG_RUN_MODE") or "").strip().lower() mode = (os.getenv("JIANGCHANG_RUN_MODE") or "").strip().lower()
if mode in ("job", "cli"): if mode in ("job", "cli"):
@@ -54,6 +82,12 @@ def get_job_id() -> str:
return _cached_job_id 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: def _get_journal_path() -> str:
global _journal_path global _journal_path
if _journal_path is None: if _journal_path is None:
@@ -93,6 +127,103 @@ def _now_ms() -> int:
return int(time.time() * 1000) 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( def _build_event(
type_: str, type_: str,
text: str, text: str,
@@ -130,7 +261,7 @@ def _build_event(
return event return event
def emit( def _emit_core(
text: str, text: str,
*, *,
type: str = "activity", type: str = "activity",
@@ -141,7 +272,6 @@ def emit(
item_id: str | None = None, item_id: str | None = None,
**extra: Any, **extra: Any,
) -> None: ) -> None:
"""Append one SAS event to the Run Journal. Never writes to stdout."""
if type == "debug" and not _activity_debug_enabled(): if type == "debug" and not _activity_debug_enabled():
return return
if type not in _VALID_EMIT_TYPES: if type not in _VALID_EMIT_TYPES:
@@ -159,11 +289,67 @@ def emit(
_write_journal_line(event) _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: def step(text: str, **kwargs: Any) -> None:
"""Shorthand for emit(type='activity').""" """Shorthand for emit(type='activity')."""
emit(text, type="activity", **kwargs) 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: def _serialize_result(result: dict[str, Any]) -> str:
line = json.dumps(result, ensure_ascii=False, separators=(",", ":")) line = json.dumps(result, ensure_ascii=False, separators=(",", ":"))
if len(line) <= _RESULT_MAX_CHARS: if len(line) <= _RESULT_MAX_CHARS:
@@ -269,6 +455,14 @@ def job_context(
try: try:
yield yield
except JobStopped as exc:
finish(
status="failed",
message=str(exc) or "用户停止",
skill=skill,
error_code="JOB_STOPPED",
)
raise
except Exception as exc: except Exception as exc:
finish(status="failed", message=str(exc), skill=skill) finish(status="failed", message=str(exc), skill=skill)
raise raise
@@ -293,7 +487,7 @@ def job_context(
def rpa_step(name: str | None = None) -> Callable[[F], F]: 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: def decorator(fn: F) -> F:
step_name = name or fn.__name__ step_name = name or fn.__name__
@@ -302,26 +496,32 @@ def rpa_step(name: str | None = None) -> Callable[[F], F]:
@functools.wraps(fn) @functools.wraps(fn)
async def async_wrapper(*args: Any, **kwargs: Any) -> Any: async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
emit(f"{step_name}") _step_gate(step_name)
_emit_core(f"{step_name}")
try: try:
result = await fn(*args, **kwargs) result = await fn(*args, **kwargs)
emit(f"{step_name}") _emit_core(f"{step_name}")
return result return result
except JobStopped:
raise
except Exception as exc: except Exception as exc:
emit(f"{step_name}: {exc}", type="warn") _emit_core(f"{step_name}: {exc}", type="warn")
raise raise
return async_wrapper # type: ignore[return-value] return async_wrapper # type: ignore[return-value]
@functools.wraps(fn) @functools.wraps(fn)
def sync_wrapper(*args: Any, **kwargs: Any) -> Any: def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
emit(f"{step_name}") _step_gate(step_name)
_emit_core(f"{step_name}")
try: try:
result = fn(*args, **kwargs) result = fn(*args, **kwargs)
emit(f"{step_name}") _emit_core(f"{step_name}")
return result return result
except JobStopped:
raise
except Exception as exc: except Exception as exc:
emit(f"{step_name}: {exc}", type="warn") _emit_core(f"{step_name}: {exc}", type="warn")
raise raise
return sync_wrapper # type: ignore[return-value] 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: 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 _cached_job_id, _journal_path, _meta_written, _journal_file
global _step_index, _current_step_name
if _journal_file is not None: if _journal_file is not None:
try: try:
_journal_file.close() _journal_file.close()
@@ -341,3 +542,5 @@ def _reset_for_tests() -> None:
_journal_path = None _journal_path = None
_meta_written = False _meta_written = False
_journal_file = None _journal_file = None
_step_index = 0
_current_step_name = None

View File

@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import os import os
import re
import shutil import shutil
from typing import Any from typing import Any
@@ -15,6 +16,29 @@ _user_env_cache: dict[str, str] | None = None
_example_defaults_cache: dict[str, str] | None = None _example_defaults_cache: dict[str, str] | None = None
def _parse_env_value(raw_value: str) -> str:
"""Parse one .env value: strip quotes / trailing comments; empty+# → \"\"."""
value = raw_value if raw_value is not None else ""
stripped = value.strip()
if not stripped:
return ""
# Quoted values keep interior content (including #).
if len(stripped) >= 2 and stripped[0] == stripped[-1] and stripped[0] in ('"', "'"):
return stripped[1:-1]
# Unquoted: ``KEY= # comment`` or ``KEY=# comment`` must be empty, not the comment text.
# Also strip ``KEY=value # comment``. Keep ``KEY=https://x/#y`` (no whitespace before #).
if re.search(r"(^|\s)#", value):
# Split at first whitespace-#-comment or whole-value comment.
m = re.match(r"^(.*?)\s+#.*$", value)
if m is not None:
return m.group(1).strip()
if stripped.startswith("#"):
return ""
return stripped
def _parse_env_file(path: str) -> dict[str, str]: def _parse_env_file(path: str) -> dict[str, str]:
"""标准库手写 .env 解析(不引 python-dotenv""" """标准库手写 .env 解析(不引 python-dotenv"""
result: dict[str, str] = {} result: dict[str, str] = {}
@@ -29,14 +53,9 @@ def _parse_env_file(path: str) -> dict[str, str]:
continue continue
key, _, value = line.partition("=") key, _, value = line.partition("=")
key = key.strip() key = key.strip()
value = value.strip()
if not key: if not key:
continue continue
if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"): result[key] = _parse_env_value(value)
value = value[1:-1]
elif " #" in value:
value = value.split(" #", 1)[0].strip()
result[key] = value
return result return result

View File

@@ -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()

View File

@@ -81,6 +81,49 @@ class TestConfig(unittest.TestCase):
self.assertEqual(config.get_float("BAD", 9.9), 9.9) self.assertEqual(config.get_float("BAD", 9.9), 9.9)
self.assertEqual(config.get_int("BAD", 7), 7) self.assertEqual(config.get_int("BAD", 7), 7)
def test_parse_empty_value_with_inline_comment(self) -> None:
"""Regression: ``KEY= # comment`` must be empty, not ``# comment``."""
with tempfile.TemporaryDirectory() as tmp:
example = os.path.join(tmp, ".env.example")
with open(example, "w", encoding="utf-8") as f:
f.write(
"DEFAULT_ACCOUNT_ID= # 可填账号管理中的编号\n"
"DEFAULT_LOGIN_ID=# 只填登录名\n"
"HAS_VALUE=real # trailing note\n"
"URL=https://example.com/path#frag\n"
"QUOTED=\"keep # hash\"\n"
"EMPTY=\n"
)
os.environ["JIANGCHANG_DATA_ROOT"] = tmp
os.environ["JIANGCHANG_USER_ID"] = "u1"
# Clear process overrides for these keys if any
for k in (
"DEFAULT_ACCOUNT_ID",
"DEFAULT_LOGIN_ID",
"HAS_VALUE",
"URL",
"QUOTED",
"EMPTY",
):
os.environ.pop(k, None)
dest = config.ensure_env_file("sk-comment", example)
config.reset_cache()
parsed = config._parse_env_file(dest)
self.assertEqual(parsed.get("DEFAULT_ACCOUNT_ID"), "")
self.assertEqual(parsed.get("DEFAULT_LOGIN_ID"), "")
self.assertEqual(parsed.get("HAS_VALUE"), "real")
self.assertEqual(parsed.get("URL"), "https://example.com/path#frag")
self.assertEqual(parsed.get("QUOTED"), "keep # hash")
self.assertEqual(parsed.get("EMPTY"), "")
# Empty example/user values must not surface as fake non-empty defaults.
self.assertIsNone(config.get("DEFAULT_ACCOUNT_ID"))
self.assertIsNone(config.get("DEFAULT_LOGIN_ID"))
self.assertEqual(config.get("HAS_VALUE"), "real")
def test_merge_missing_env_keys_appends_only_new(self) -> None: def test_merge_missing_env_keys_appends_only_new(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
example = os.path.join(tmp, ".env.example") example = os.path.join(tmp, ".env.example")

View File

@@ -39,9 +39,23 @@ def test_workflow_packages_readme_at_root(workflow_text: str) -> None:
def test_workflow_metadata_reads_root_readme(workflow_text: str) -> None: def test_workflow_metadata_reads_root_readme(workflow_text: str) -> None:
assert "readme_path = 'README.md'" in workflow_text assert 'load_md("README.md")' in workflow_text
assert "frontmatter.load(readme_path)" in workflow_text assert 'metadata["readme_md"] = body' in workflow_text
assert "metadata['readme_md'] = body" in workflow_text
def test_workflow_packages_market_docs_when_present(workflow_text: str) -> None:
assert 'for market_doc in ("TUTORIAL.md", "DEMO.md", "CHANGELOG.md"):' in workflow_text
assert "Copied {market_doc}" in workflow_text
def test_workflow_metadata_syncs_tutorial_demo_changelog(workflow_text: str) -> None:
assert 'load_md("TUTORIAL.md")' in workflow_text
assert 'metadata["tutorial_md"] = tutorial_body' in workflow_text
assert 'load_md("DEMO.md")' in workflow_text
assert 'metadata["demo_md"] = demo_body' in workflow_text
assert 'metadata["demo_video_url"]' in workflow_text
assert "extract_changelog_for_version" in workflow_text
assert 'metadata["changelog"] = section' in workflow_text
def test_workflow_does_not_use_references_readme_for_marketplace( def test_workflow_does_not_use_references_readme_for_marketplace(