fix(config): treat empty .env values with inline # comments as empty

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>

.github/workflows/reusable-release-skill.yaml

@@ -115,6 +115,12 @@ jobs:
             "    print('Copied README.md')" \
             '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')" \
+            '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('assets', os.path.join(PACKAGE, 'assets'), common_ignore)" \
             "copy_dir('tests', os.path.join(PACKAGE, 'tests'), common_ignore)" \
@@ -144,49 +150,119 @@ jobs:
       - name: Parse Metadata and Pack
         id: build_task
         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 {})
-          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:
-                  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():
-                      metadata['description'] = str(desc).strip()
+                      metadata["description"] = str(desc).strip()
                   if body:
-                      metadata['readme_md'] = body
+                      metadata["readme_md"] = body
               except Exception:
                   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:
-              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
         env:

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

pyproject.toml

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

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",

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

src/jiangchang_skill_core/config.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import re
 import shutil
 from typing import Any
 
@@ -15,6 +16,29 @@ _user_env_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]:
     """标准库手写 .env 解析（不引 python-dotenv）。"""
     result: dict[str, str] = {}
@@ -29,14 +53,9 @@ def _parse_env_file(path: str) -> dict[str, str]:
                 continue
             key, _, value = line.partition("=")
             key = key.strip()
-            value = value.strip()
             if not key:
                 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
 
 

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

tests/test_config.py

@@ -81,6 +81,49 @@ class TestConfig(unittest.TestCase):
             self.assertEqual(config.get_float("BAD", 9.9), 9.9)
             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:
         with tempfile.TemporaryDirectory() as tmp:
             example = os.path.join(tmp, ".env.example")

tests/test_release_workflow.py

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