feat(activity): add SAS stream_delta API for LLM progressive text (1.2.14)

Add precise Chrome/Edge quit by user-data-dir and stale SingletonLock cleanup so publish RPAs can free account-manager windows before launch without killing the default browser.

README.md

@@ -87,6 +87,8 @@ send_prompt_via_composer(page, "第一行\n第二行")
 |-----|------|
 | `emit(text, ...)` | 写 Run Journal（`activity` / `progress` / `warn` / `debug`） |
 | `step(text)` | `emit(type='activity')` 别名 |
+| `stream_start` / `stream_delta` / `stream_end` | 写 `type=stream` 渐进文本（**不**步进闸门；供 HTML `jiangchang.stream`） |
+| `stream_channel(...)` | 上下文管理器：start → delta → end |
 | `finish(status=..., **fields)` | 写终态事件 + **唯一**向 stdout 输出单行 result JSON |
 | `rpa_step(name)` | 装饰器：闸门 + 进入/成功/失败自动 emit |
 | `checkpoint()` | consult control（不写步骤事件） |
@@ -94,6 +96,21 @@ send_prompt_via_composer(page, "第一行\n第二行")
 | `JobStopped` | 宿主 stop 时抛出 |
 | `job_context(...)` | 可选包裹 main；异常时自动 `finish(failed)` |
 
+### LLM 流式示例（1.2.14+）
+
+```python
+from jiangchang_skill_core.activity import finish, stream_channel
+
+with stream_channel("answer", skill="reply-xmz-knowledge") as ch:
+    for piece in llm_stream(...):  # OpenAI/DeepSeek stream chunks
+        ch.delta(piece)
+
+finish(status="success", skill="reply-xmz-knowledge", message=full_text, answer=full_text)
+```
+
+- `stream_delta` 默认合并短突发（约 50ms / 48 字符）；可用环境变量 `JIANGCHANG_STREAM_COALESCE_MS` / `JIANGCHANG_STREAM_COALESCE_CHARS` 调整，或 `flush=True` 立即写出。
+- 宿主 Task Center **不展示**逐字 `stream` delta；HTML 预览通过 `jiangchang.stream` 订阅。
+
 ### Run Journal 路径（宿主 tail 此文件）
 
 ```
@@ -278,8 +295,12 @@ print(runtime_diagnostics_dict(diag))  # 可 JSON 序列化
 
 诊断只读探测本地状态，不会触发 media-assets 下载。
 
+自 **1.2.12** 起，统一文件日志 `jiangchang.log` 的时间戳为 `YYYY-MM-DD HH:MM:SS`（空格分隔，不再使用 ISO 的 `T`）。宿主运行日志解析需同时兼容新旧两种格式。
+
 自 **1.2.11** 起，`runtime_diagnostics_dict` 额外包含只读字段 `machine_power`（AC 供电 / 可选 nvidia-smi 限频信息）。探不到则 `probed=false`，**不会**因此把 health 标为 fatal。
 
+自 **1.2.13** 起，提供按 Profile 精准关闭：`ensure_profile_free` / `force_quit_profile_browsers`（只杀命令行含该 `user-data-dir` 的 Chrome/Edge，不碰系统默认浏览器），并可选清理崩溃残留 `SingletonLock`。
+
 ### RPA 有界就绪等待与启动超时（1.2.11+）
 
 ```python
@@ -296,6 +317,20 @@ if result.timed_out:
 # 或环境变量 OPENCLAW_BROWSER_LAUNCH_TIMEOUT_MS=60000
 ```
 
+### RPA Profile 占用释放（1.2.13+）
+
+```python
+from jiangchang_skill_core.rpa import ensure_profile_free, launch_persistent_browser
+from jiangchang_skill_core.rpa.errors import ERR_PROFILE_IN_USE
+
+# 发布前：关掉账号管理「打开浏览器」占用的同一 Profile，再启动
+result = ensure_profile_free(profile_dir)
+# result.pids_before / killed_ops / stale_lock_cleared / freed
+
+# 或在统一启动封装上打开开关（默认 False，保持旧行为）
+# context = await launch_persistent_browser(pw, profile_dir=..., ensure_free=True)
+```
+
 薄错误码（技能自行决定是否 fatal / retryable）：
 
 - `ERROR:UI_NOT_READY`

pyproject.toml

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

src/jiangchang_skill_core/__init__.py

@@ -1,5 +1,6 @@
 from .activity import (
     JobStopped,
+    StreamChannel,
     checkpoint,
     emit,
     finish,
@@ -11,6 +12,10 @@ from .activity import (
     job_context,
     rpa_step,
     step,
+    stream_channel,
+    stream_delta,
+    stream_end,
+    stream_start,
 )
 from .sibling_bridge import (
     call_sibling_json,
@@ -74,6 +79,7 @@ except ImportError:
 
 __all__ = [
     "JobStopped",
+    "StreamChannel",
     "call_sibling_json",
     "call_sibling_json_array",
     "checkpoint",
@@ -89,6 +95,10 @@ __all__ = [
     "job_context",
     "rpa_step",
     "step",
+    "stream_channel",
+    "stream_delta",
+    "stream_end",
+    "stream_start",
     "get_sibling_main_path",
     "EntitlementClient",
     "EntitlementDeniedError",

src/jiangchang_skill_core/activity.py

@@ -4,6 +4,9 @@ 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.
 
+Stream helpers (stream_start / stream_delta / stream_end / stream_channel) write
+type=stream journal events without advancing the step gate — for LLM token UX.
+
 SRCP: host writes {job_id}.control.json; kit consults at step gates (emit,
 rpa_step, interruptible_sleep) for pause / resume / stop.
 """
@@ -26,9 +29,13 @@ 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_STREAM_PHASES = frozenset({"start", "delta", "end"})
 _VALID_CONTROL_COMMANDS = frozenset({"none", "pause", "resume", "stop"})
 _RESULT_MAX_CHARS = 4096
 _DEFAULT_CONTROL_POLL_SEC = 0.2
+_DEFAULT_STREAM_COALESCE_MS = 50
+_DEFAULT_STREAM_COALESCE_CHARS = 48
+_STREAM_CHECKPOINT_EVERY_MS = 200
 
 _cached_job_id: Optional[str] = None
 _journal_path: Optional[str] = None
@@ -36,6 +43,8 @@ _meta_written: bool = False
 _journal_file: Optional[Any] = None
 _step_index: int = 0
 _current_step_name: Optional[str] = None
+_stream_last_checkpoint_ms: int = 0
+_stream_buffers: dict[str, dict[str, Any]] = {}
 
 F = TypeVar("F", bound=Callable[..., Any])
 
@@ -320,6 +329,220 @@ def step(text: str, **kwargs: Any) -> None:
     emit(text, type="activity", **kwargs)
 
 
+def _stream_coalesce_ms() -> int:
+    raw = (os.getenv("JIANGCHANG_STREAM_COALESCE_MS") or "").strip()
+    if not raw:
+        return _DEFAULT_STREAM_COALESCE_MS
+    try:
+        return max(0, int(raw))
+    except ValueError:
+        return _DEFAULT_STREAM_COALESCE_MS
+
+
+def _stream_coalesce_chars() -> int:
+    raw = (os.getenv("JIANGCHANG_STREAM_COALESCE_CHARS") or "").strip()
+    if not raw:
+        return _DEFAULT_STREAM_COALESCE_CHARS
+    try:
+        return max(1, int(raw))
+    except ValueError:
+        return _DEFAULT_STREAM_COALESCE_CHARS
+
+
+def _maybe_stream_checkpoint() -> None:
+    """Consult stop/pause without advancing step index (throttled)."""
+    global _stream_last_checkpoint_ms
+    now = _now_ms()
+    if now - _stream_last_checkpoint_ms < _STREAM_CHECKPOINT_EVERY_MS:
+        return
+    _stream_last_checkpoint_ms = now
+    checkpoint()
+
+
+def _write_stream_event(
+    *,
+    phase: str,
+    stream_id: str,
+    text: str,
+    delta: str | None = None,
+    skill: str | None = None,
+    **extra: Any,
+) -> None:
+    if phase not in _VALID_STREAM_PHASES:
+        raise ValueError(f"invalid stream phase: {phase!r}")
+    sid = (stream_id or "answer").strip() or "answer"
+    label = (text or "").strip() or sid
+    event = _build_event(
+        "stream",
+        label,
+        skill=skill,
+        phase=phase,
+        stream_id=sid,
+        **extra,
+    )
+    if delta is not None:
+        event["delta"] = delta
+    _write_journal_line(event)
+
+
+def stream_start(
+    stream_id: str = "answer",
+    *,
+    skill: str | None = None,
+    text: str | None = None,
+    **extra: Any,
+) -> None:
+    """Open a progressive text stream (no step gate)."""
+    _maybe_stream_checkpoint()
+    sid = (stream_id or "answer").strip() or "answer"
+    buf = _stream_buffers.get(sid)
+    if buf and buf.get("pending"):
+        _flush_stream_buffer(sid, force=True)
+    _stream_buffers[sid] = {
+        "pending": "",
+        "last_flush_ms": _now_ms(),
+        "skill": skill,
+        "extra": dict(extra),
+    }
+    _write_stream_event(
+        phase="start",
+        stream_id=sid,
+        text=text or f"stream:{sid}",
+        skill=skill,
+        **extra,
+    )
+
+
+def _flush_stream_buffer(stream_id: str, *, force: bool = False) -> None:
+    buf = _stream_buffers.get(stream_id)
+    if not buf:
+        return
+    pending = buf.get("pending") or ""
+    if not pending:
+        return
+    coalesce_ms = _stream_coalesce_ms()
+    coalesce_chars = _stream_coalesce_chars()
+    now = _now_ms()
+    aged = now - int(buf.get("last_flush_ms") or 0) >= coalesce_ms
+    big = len(pending) >= coalesce_chars
+    if not force and not aged and not big:
+        return
+    _maybe_stream_checkpoint()
+    skill = buf.get("skill")
+    extra = dict(buf.get("extra") or {})
+    _write_stream_event(
+        phase="delta",
+        stream_id=stream_id,
+        text=pending,
+        delta=pending,
+        skill=skill if isinstance(skill, str) else None,
+        **extra,
+    )
+    buf["pending"] = ""
+    buf["last_flush_ms"] = now
+
+
+def stream_delta(
+    delta: str,
+    *,
+    stream_id: str = "answer",
+    skill: str | None = None,
+    flush: bool = False,
+    **extra: Any,
+) -> None:
+    """
+    Append progressive text to the Run Journal as type=stream phase=delta.
+
+    Does not advance the SAS step index. Empty deltas are ignored.
+    Small chunks are coalesced (~50ms / ~48 chars) unless flush=True.
+    """
+    piece = delta if isinstance(delta, str) else str(delta or "")
+    if piece == "":
+        return
+    sid = (stream_id or "answer").strip() or "answer"
+    buf = _stream_buffers.get(sid)
+    if buf is None:
+        stream_start(sid, skill=skill, **extra)
+        buf = _stream_buffers.get(sid)
+    assert buf is not None
+    if skill is not None:
+        buf["skill"] = skill
+    if extra:
+        merged = dict(buf.get("extra") or {})
+        merged.update(extra)
+        buf["extra"] = merged
+    buf["pending"] = (buf.get("pending") or "") + piece
+    _flush_stream_buffer(sid, force=flush)
+
+
+def stream_end(
+    stream_id: str = "answer",
+    *,
+    skill: str | None = None,
+    text: str | None = None,
+    **extra: Any,
+) -> None:
+    """Flush pending deltas and close a progressive text stream."""
+    sid = (stream_id or "answer").strip() or "answer"
+    _flush_stream_buffer(sid, force=True)
+    buf = _stream_buffers.pop(sid, None)
+    use_skill = skill
+    if use_skill is None and buf and isinstance(buf.get("skill"), str):
+        use_skill = buf["skill"]
+    merged_extra = dict(buf.get("extra") or {}) if buf else {}
+    merged_extra.update(extra)
+    _maybe_stream_checkpoint()
+    _write_stream_event(
+        phase="end",
+        stream_id=sid,
+        text=text or f"stream:{sid}:end",
+        skill=use_skill,
+        **merged_extra,
+    )
+
+
+class StreamChannel:
+    """Context manager around stream_start / stream_delta / stream_end."""
+
+    def __init__(
+        self,
+        stream_id: str = "answer",
+        *,
+        skill: str | None = None,
+        **extra: Any,
+    ) -> None:
+        self.stream_id = (stream_id or "answer").strip() or "answer"
+        self.skill = skill
+        self.extra = dict(extra)
+
+    def __enter__(self) -> "StreamChannel":
+        stream_start(self.stream_id, skill=self.skill, **self.extra)
+        return self
+
+    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
+        stream_end(self.stream_id, skill=self.skill)
+        return None
+
+    def delta(self, text: str, *, flush: bool = False) -> None:
+        stream_delta(
+            text,
+            stream_id=self.stream_id,
+            skill=self.skill,
+            flush=flush,
+            **self.extra,
+        )
+
+
+def stream_channel(
+    stream_id: str = "answer",
+    *,
+    skill: str | None = None,
+    **extra: Any,
+) -> StreamChannel:
+    """Return a StreamChannel context manager for progressive LLM text."""
+    return StreamChannel(stream_id, skill=skill, **extra)
+
+
 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:
@@ -532,7 +755,7 @@ def rpa_step(name: str | None = None) -> Callable[[F], F]:
 def _reset_for_tests() -> None:
     """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
+    global _step_index, _current_step_name, _stream_last_checkpoint_ms, _stream_buffers
     if _journal_file is not None:
         try:
             _journal_file.close()
@@ -544,3 +767,5 @@ def _reset_for_tests() -> None:
     _journal_file = None
     _step_index = 0
     _current_step_name = None
+    _stream_last_checkpoint_ms = 0
+    _stream_buffers = {}

src/jiangchang_skill_core/rpa/__init__.py

@@ -14,6 +14,15 @@ try:
 except ImportError:
     RpaVideoSession = None  # type: ignore[misc, assignment]
 
+from .profile_lock import (
+    EnsureProfileFreeResult,
+    clear_stale_singleton_locks,
+    ensure_profile_free,
+    force_quit_profile_browsers,
+    is_profile_lock_error,
+    list_profile_browser_pids,
+    list_profile_browser_root_pids,
+)
 from .wait import PollUntilResult, poll_until, poll_until_sync
 
 __all__ = [
@@ -28,6 +37,13 @@ __all__ = [
     "PollUntilResult",
     "poll_until",
     "poll_until_sync",
+    "EnsureProfileFreeResult",
+    "clear_stale_singleton_locks",
+    "ensure_profile_free",
+    "force_quit_profile_browsers",
+    "is_profile_lock_error",
+    "list_profile_browser_pids",
+    "list_profile_browser_root_pids",
     # re-export common anti_detect helpers
     "random_delay",
     "human_delay_short",

src/jiangchang_skill_core/rpa/browser.py

@@ -44,6 +44,7 @@ async def launch_persistent_browser(
     headless: bool | None = None,
     extra_args: list[str] | None = None,
     timeout_ms: int | None = None,
+    ensure_free: bool = False,
 ) -> "BrowserContext":
     """用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。
 
@@ -54,10 +55,20 @@ async def launch_persistent_browser(
         可选启动超时（毫秒）。为 None 时读 OPENCLAW_BROWSER_LAUNCH_TIMEOUT_MS；
         仍为空则不包裹超时（与历史行为一致）。超时抛 RuntimeError，前缀
         ERROR:BROWSER_START_TIMEOUT。本函数不做自动重试。
+
+    ensure_free:
+        为 True 时，启动前调用 ``ensure_profile_free``：精准关闭占用该
+        Profile 的 Chrome/Edge，并清理崩溃残留 Singleton 锁。默认 False，
+        保持历史行为；发布类技能建议传 True，或自行先调 ensure_profile_free。
     """
     if headless is None:
         headless = config.get_bool("OPENCLAW_BROWSER_HEADLESS", default=False)
 
+    if ensure_free:
+        from .profile_lock import ensure_profile_free
+
+        ensure_profile_free(profile_dir)
+
     chrome = executable_path or find_chrome_executable()
     if not chrome:
         raise RuntimeError(

src/jiangchang_skill_core/rpa/errors.py

@@ -12,6 +12,7 @@ ERR_WINDOW_NOT_FOUND = "ERROR:WINDOW_NOT_FOUND"
 ERR_UI_NOT_READY = "ERROR:UI_NOT_READY"
 ERR_BROWSER_START_TIMEOUT = "ERROR:BROWSER_START_TIMEOUT"
 ERR_ENV_DEGRADED = "ERROR:ENV_DEGRADED"
+ERR_PROFILE_IN_USE = "ERROR:PROFILE_IN_USE"
 
 __all__ = [
     "ERR_REQUIRE_LOGIN",
@@ -24,4 +25,5 @@ __all__ = [
     "ERR_UI_NOT_READY",
     "ERR_BROWSER_START_TIMEOUT",
     "ERR_ENV_DEGRADED",
+    "ERR_PROFILE_IN_USE",
 ]

src/jiangchang_skill_core/rpa/profile_lock.py

@@ -0,0 +1,384 @@
+"""按 Chromium Profile（user-data-dir）精准占用检测 / 关闭 / 僵锁清理。
+
+只匹配命令行包含该 Profile 绝对路径的 chrome/msedge，不碰系统默认浏览器。
+供发布类 RPA 在 launch_persistent_context 前调用，避免「账号管理打开浏览器未关」撞锁失败。
+"""
+
+from __future__ import annotations
+
+import base64
+import logging
+import os
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, field
+from typing import Iterable, List, Sequence
+
+from .errors import ERR_PROFILE_IN_USE
+
+_logger = logging.getLogger(__name__)
+
+# 过短路径误匹配面过大（正式账号 Profile 远长于此）
+_MIN_PROFILE_PATH_LEN = 12
+
+_LOCK_HINTS = (
+    "processsingleton",
+    "profile is already in use",
+    "user data directory is already in use",
+    "singletonlock",
+    "browser is already running",
+    "failed to create a processsingleton",
+)
+
+_SINGLETON_NAMES = ("SingletonLock", "SingletonCookie", "SingletonSocket")
+
+
+@dataclass(frozen=True)
+class EnsureProfileFreeResult:
+    """ensure_profile_free 的结构化结果（便于技能打日志）。"""
+
+    profile_dir: str
+    pids_before: tuple[int, ...] = ()
+    killed_ops: int = 0
+    stale_lock_cleared: bool = False
+    pids_after: tuple[int, ...] = ()
+    freed: bool = True
+    notes: tuple[str, ...] = field(default_factory=tuple)
+
+
+def is_profile_lock_error(exc: BaseException) -> bool:
+    """判断异常文案是否像 Chromium Profile 被占用。"""
+    msg = str(exc).lower()
+    return any(n in msg for n in _LOCK_HINTS)
+
+
+def _norm_profile(profile_dir: str) -> str:
+    return os.path.abspath((profile_dir or "").strip())
+
+
+def _profile_path_b64(profile_dir: str) -> str:
+    """UTF-8 Base64，供 PowerShell 解码；避免 -Command 引号/编码弄坏中文路径。"""
+    return base64.b64encode(_norm_profile(profile_dir).encode("utf-8")).decode("ascii")
+
+
+def _ps_decode_profile_path(path_b64: str) -> str:
+    return (
+        "$ErrorActionPreference = 'SilentlyContinue'\n"
+        f"$p = [Text.Encoding]::UTF8.GetString([Convert]::FromBase64String('{path_b64}'))\n"
+        "if (-not $p) { exit 0 }\n"
+    )
+
+
+def _dedupe_pids(pids: Iterable[int]) -> list[int]:
+    seen: set[int] = set()
+    out: list[int] = []
+    for pid in pids:
+        if pid <= 0 or pid in seen:
+            continue
+        seen.add(pid)
+        out.append(pid)
+    return out
+
+
+def list_profile_browser_pids(profile_dir: str) -> list[int]:
+    """查找占用该 user-data-dir 的 chrome/msedge 进程（匹配完整 Profile 路径）。"""
+    profile_abs = _norm_profile(profile_dir)
+    if not profile_abs or len(profile_abs) < _MIN_PROFILE_PATH_LEN:
+        return []
+    if sys.platform == "win32":
+        return _list_pids_win(profile_abs, roots_only=False)
+    return _list_pids_posix(profile_abs, roots_only=False)
+
+
+def list_profile_browser_root_pids(profile_dir: str) -> list[int]:
+    """仅主进程（命令行不含 --type=），便于 taskkill /T 收整棵进程树。"""
+    profile_abs = _norm_profile(profile_dir)
+    if not profile_abs or len(profile_abs) < _MIN_PROFILE_PATH_LEN:
+        return []
+    if sys.platform == "win32":
+        return _list_pids_win(profile_abs, roots_only=True)
+    return _list_pids_posix(profile_abs, roots_only=True)
+
+
+def _list_pids_win(profile_abs: str, *, roots_only: bool) -> list[int]:
+    root_clause = (
+        "$_.CommandLine -notmatch '--type='"
+        if roots_only
+        else "$true"
+    )
+    ps = (
+        _ps_decode_profile_path(_profile_path_b64(profile_abs))
+        + "Get-CimInstance Win32_Process -Filter \"Name='chrome.exe' OR Name='msedge.exe'\" |\n"
+        + "  Where-Object {\n"
+        + "    $_.CommandLine -and\n"
+        + "    $_.CommandLine.Contains($p) -and\n"
+        + f"    ({root_clause})\n"
+        + "  } |\n"
+        + "  ForEach-Object { $_.ProcessId }\n"
+    )
+    try:
+        out = subprocess.check_output(
+            ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps],
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=20,
+        )
+        return _dedupe_pids(
+            int(line.strip()) for line in (out or "").splitlines() if line.strip().isdigit()
+        )
+    except Exception:
+        _logger.warning(
+            "profile_pid_scan_failed profile_dir=%s roots_only=%s",
+            profile_abs,
+            roots_only,
+            exc_info=True,
+        )
+        return []
+
+
+def _list_pids_posix(profile_abs: str, *, roots_only: bool) -> list[int]:
+    """尽力而为：扫 chrome/chromium/msedge 命令行。无 psutil 依赖。"""
+    try:
+        out = subprocess.check_output(
+            ["ps", "-ax", "-o", "pid=,command="],
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=15,
+        )
+    except Exception:
+        _logger.warning("profile_pid_scan_failed_posix profile_dir=%s", profile_abs, exc_info=True)
+        return []
+    needle = profile_abs
+    names = ("chrome", "chromium", "msedge", "google-chrome")
+    pids: list[int] = []
+    for line in (out or "").splitlines():
+        text = line.strip()
+        if not text:
+            continue
+        parts = text.split(None, 1)
+        if len(parts) < 2:
+            continue
+        try:
+            pid = int(parts[0])
+        except ValueError:
+            continue
+        cmd = parts[1]
+        if needle not in cmd:
+            continue
+        low = cmd.lower()
+        if not any(n in low for n in names):
+            continue
+        if roots_only and "--type=" in cmd:
+            continue
+        pids.append(pid)
+    return _dedupe_pids(pids)
+
+
+def force_quit_profile_browsers(profile_dir: str, *, settle_sec: float = 0.4) -> int:
+    """仅结束命令行包含该 Profile 绝对路径的 Chrome/Edge 进程树。
+
+    返回发起结束的次数（含二次清残留），不表示最终仍占用的进程数。
+    """
+    raw = (profile_dir or "").strip()
+    if not raw:
+        _logger.warning("browser_force_quit_skipped_bad_profile profile_dir=%r", profile_dir)
+        return 0
+    profile_abs = _norm_profile(raw)
+    if len(profile_abs) < _MIN_PROFILE_PATH_LEN:
+        _logger.warning("browser_force_quit_skipped_bad_profile profile_dir=%r", profile_dir)
+        return 0
+
+    killed = 0
+    roots = list_profile_browser_root_pids(profile_abs)
+    targets = roots or list_profile_browser_pids(profile_abs)
+    killed += _kill_pids(targets)
+
+    if settle_sec > 0:
+        time.sleep(settle_sec)
+    leftovers = list_profile_browser_pids(profile_abs)
+    if leftovers:
+        killed += _kill_pids(leftovers, force=True)
+
+    remain = list_profile_browser_pids(profile_abs)
+    _logger.info(
+        "browser_force_quit profile_dir=%s roots=%s killed_ops=%s remain=%s",
+        profile_abs,
+        roots,
+        killed,
+        remain,
+    )
+    return killed
+
+
+def _kill_pids(pids: Sequence[int], *, force: bool = False) -> int:
+    ops = 0
+    for pid in pids:
+        try:
+            if sys.platform == "win32":
+                subprocess.run(
+                    ["taskkill", "/PID", str(pid), "/T", "/F"],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                    check=False,
+                )
+            else:
+                os.kill(pid, 9 if force else 15)
+            ops += 1
+        except Exception:
+            _logger.warning("browser_force_quit_pid_failed pid=%s", pid, exc_info=True)
+    return ops
+
+
+def _pid_alive(pid: int) -> bool:
+    if pid <= 0:
+        return False
+    try:
+        if sys.platform == "win32":
+            # 0 信号在 Windows 上不可靠；用 tasklist
+            out = subprocess.check_output(
+                ["tasklist", "/FI", f"PID eq {pid}", "/NH"],
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                timeout=10,
+            )
+            return str(pid) in (out or "")
+        os.kill(pid, 0)
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    except Exception:
+        return False
+
+
+def _parse_singleton_target_pid(lock_path: str) -> int | None:
+    """从 SingletonLock 解析目标 PID（symlink 目标形如 hostname-pid）。"""
+    try:
+        if os.path.islink(lock_path):
+            target = os.readlink(lock_path)
+        else:
+            # Windows 上偶发为普通文件，内容也可能是 hostname-pid
+            with open(lock_path, "rb") as fh:
+                raw = fh.read(256)
+            target = raw.decode("utf-8", errors="ignore").strip()
+            if not target:
+                return None
+        pid_s = target.rsplit("-", 1)[-1].strip()
+        pid = int(pid_s)
+        return pid if pid > 0 else None
+    except (OSError, ValueError):
+        return None
+
+
+def clear_stale_singleton_locks(profile_dir: str) -> bool:
+    """若 Profile 无存活占用进程，清理崩溃残留的 Singleton* 文件。
+
+    有活进程占用时不删。锁指向的 PID 仍存活时也不删。
+    返回是否删除了至少一个文件。
+    """
+    profile_abs = _norm_profile(profile_dir)
+    if not profile_abs or not os.path.isdir(profile_abs):
+        return False
+    if list_profile_browser_pids(profile_abs):
+        return False
+
+    lock_path = os.path.join(profile_abs, "SingletonLock")
+    if os.path.lexists(lock_path):
+        pid = _parse_singleton_target_pid(lock_path)
+        if pid is not None and _pid_alive(pid):
+            # 锁指向的进程仍在，但命令行未必含本路径（少见）——保守不删
+            _logger.info(
+                "singleton_lock_held_by_pid profile_dir=%s pid=%s",
+                profile_abs,
+                pid,
+            )
+            return False
+
+    removed = False
+    for name in _SINGLETON_NAMES:
+        path = os.path.join(profile_abs, name)
+        if not os.path.lexists(path):
+            continue
+        try:
+            os.remove(path)
+            removed = True
+            _logger.info("stale_singleton_removed profile_dir=%s name=%s", profile_abs, name)
+        except OSError:
+            _logger.warning(
+                "stale_singleton_remove_failed profile_dir=%s name=%s",
+                profile_abs,
+                name,
+                exc_info=True,
+            )
+    return removed
+
+
+def ensure_profile_free(
+    profile_dir: str,
+    *,
+    force_quit: bool = True,
+    clear_stale_lock: bool = True,
+    wait_sec: float = 1.0,
+    raise_if_busy: bool = True,
+) -> EnsureProfileFreeResult:
+    """确保指定 Profile 可被 Playwright launch_persistent_context 独占使用。
+
+    默认：精准关闭占用该目录的浏览器 → 清僵锁 → 复查。
+    仍被占用且 raise_if_busy=True 时抛 RuntimeError（前缀 ERROR:PROFILE_IN_USE）。
+    """
+    profile_abs = _norm_profile(profile_dir)
+    notes: List[str] = []
+    if not profile_abs:
+        raise ValueError("profile_dir 不能为空")
+    if len(profile_abs) < _MIN_PROFILE_PATH_LEN:
+        raise ValueError(f"profile_dir 过短，拒绝操作: {profile_abs!r}")
+
+    before = tuple(list_profile_browser_pids(profile_abs))
+    killed_ops = 0
+    if before and force_quit:
+        killed_ops = force_quit_profile_browsers(profile_abs)
+        notes.append(f"force_quit killed_ops={killed_ops}")
+    elif before and not force_quit:
+        notes.append("occupied_skip_quit")
+
+    stale_cleared = False
+    if clear_stale_lock:
+        stale_cleared = clear_stale_singleton_locks(profile_abs)
+        if stale_cleared:
+            notes.append("stale_singleton_cleared")
+
+    if wait_sec > 0:
+        time.sleep(wait_sec)
+
+    after = tuple(list_profile_browser_pids(profile_abs))
+    freed = len(after) == 0
+    result = EnsureProfileFreeResult(
+        profile_dir=profile_abs,
+        pids_before=before,
+        killed_ops=killed_ops,
+        stale_lock_cleared=stale_cleared,
+        pids_after=after,
+        freed=freed,
+        notes=tuple(notes),
+    )
+    _logger.info(
+        "ensure_profile_free profile_dir=%s before=%s after=%s killed_ops=%s "
+        "stale_cleared=%s freed=%s",
+        profile_abs,
+        before,
+        after,
+        killed_ops,
+        stale_cleared,
+        freed,
+    )
+    if not freed and raise_if_busy:
+        raise RuntimeError(
+            f"{ERR_PROFILE_IN_USE} 账号浏览器配置目录仍被占用（pids={list(after)}）："
+            f"{profile_abs}。请关闭账号管理打开的对应窗口后重试。"
+        )
+    return result

src/jiangchang_skill_core/unified_logging.py

@@ -87,7 +87,7 @@ class _SkillContextFilter(logging.Filter):
 _FORMAT = (
     "%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
 )
-_DATEFMT = "%Y-%m-%dT%H:%M:%S"
+_DATEFMT = "%Y-%m-%d %H:%M:%S"
 
 
 def _log_level_from_env() -> int:

tests/test_activity_stream.py

@@ -0,0 +1,144 @@
+# -*- coding: utf-8 -*-
+"""SAS stream_start / stream_delta / stream_end / stream_channel."""
+from __future__ import annotations
+
+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 _ActivityEnvIsolation(unittest.TestCase):
+    _KEYS = (
+        "JIANGCHANG_DATA_ROOT",
+        "JIANGCHANG_USER_ID",
+        "JIANGCHANG_JOB_ID",
+        "JIANGCHANG_RUN_MODE",
+        "JIANGCHANG_ACTIVITY_DEBUG",
+        "JIANGCHANG_STREAM_COALESCE_MS",
+        "JIANGCHANG_STREAM_COALESCE_CHARS",
+        "JIANGCHANG_CONTROL_DISABLED",
+    )
+
+    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)
+        os.environ["JIANGCHANG_CONTROL_DISABLED"] = "1"
+        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 _events(self, job_id: str) -> list[dict]:
+        journal = Path(get_runs_dir()) / f"{job_id}.jsonl"
+        lines = journal.read_text(encoding="utf-8").strip().splitlines()
+        return [json.loads(line) for line in lines]
+
+
+class TestActivityStream(_ActivityEnvIsolation):
+    def test_stream_writes_start_delta_end_without_step_gate(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-stream-1"
+        os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0"
+        activity._reset_for_tests()
+
+        activity.stream_start("answer", skill="demo")
+        activity.stream_delta("你", stream_id="answer", skill="demo")
+        activity.stream_delta("好", stream_id="answer", skill="demo")
+        activity.stream_end("answer", skill="demo")
+
+        events = self._events("job-stream-1")
+        stream_events = [e for e in events if e.get("type") == "stream"]
+        phases = [e.get("phase") for e in stream_events]
+        self.assertEqual(phases[0], "start")
+        self.assertEqual(phases[-1], "end")
+        deltas = [e.get("delta") for e in stream_events if e.get("phase") == "delta"]
+        self.assertEqual("".join(deltas), "你好")
+        self.assertEqual(activity._step_index, 0)
+
+    def test_stream_channel_context_manager(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-stream-ch"
+        os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0"
+        activity._reset_for_tests()
+
+        with activity.stream_channel("answer", skill="demo") as ch:
+            ch.delta("A")
+            ch.delta("B", flush=True)
+
+        events = self._events("job-stream-ch")
+        stream_events = [e for e in events if e.get("type") == "stream"]
+        self.assertEqual(stream_events[0]["phase"], "start")
+        self.assertEqual(stream_events[-1]["phase"], "end")
+        deltas = [e.get("delta") for e in stream_events if e.get("phase") == "delta"]
+        self.assertEqual("".join(d for d in deltas if d), "AB")
+
+    def test_stream_delta_auto_starts(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-stream-auto"
+        os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0"
+        activity._reset_for_tests()
+
+        activity.stream_delta("x", stream_id="answer")
+        activity.stream_end("answer")
+
+        events = self._events("job-stream-auto")
+        stream_events = [e for e in events if e.get("type") == "stream"]
+        self.assertEqual(stream_events[0]["phase"], "start")
+        self.assertTrue(any(e.get("phase") == "delta" and e.get("delta") == "x" for e in stream_events))
+
+    def test_empty_delta_ignored(self) -> None:
+        tmp = self._tmp_data_root()
+        os.environ["JIANGCHANG_DATA_ROOT"] = tmp
+        os.environ["JIANGCHANG_JOB_ID"] = "job-stream-empty"
+        os.environ["JIANGCHANG_STREAM_COALESCE_MS"] = "0"
+        activity._reset_for_tests()
+
+        activity.stream_start("answer")
+        activity.stream_delta("", stream_id="answer")
+        activity.stream_end("answer")
+
+        events = self._events("job-stream-empty")
+        stream_events = [e for e in events if e.get("type") == "stream"]
+        self.assertFalse(any(e.get("phase") == "delta" for e in stream_events))
+
+    def test_public_exports(self) -> None:
+        from jiangchang_skill_core import (
+            StreamChannel,
+            stream_channel,
+            stream_delta,
+            stream_end,
+            stream_start,
+        )
+
+        self.assertTrue(callable(stream_start))
+        self.assertTrue(callable(stream_delta))
+        self.assertTrue(callable(stream_end))
+        self.assertTrue(callable(stream_channel))
+        self.assertTrue(issubclass(StreamChannel, object))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_profile_lock.py

@@ -0,0 +1,134 @@
+# -*- coding: utf-8 -*-
+"""Profile 精准关闭 / 僵锁清理单元测试。"""
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from jiangchang_skill_core.rpa.errors import ERR_PROFILE_IN_USE
+from jiangchang_skill_core.rpa.profile_lock import (
+    clear_stale_singleton_locks,
+    ensure_profile_free,
+    force_quit_profile_browsers,
+    is_profile_lock_error,
+)
+
+
+class TestProfileLockHelpers(unittest.TestCase):
+    def test_is_profile_lock_error(self) -> None:
+        self.assertTrue(
+            is_profile_lock_error(
+                RuntimeError("User data directory is already in use")
+            )
+        )
+        self.assertTrue(is_profile_lock_error(RuntimeError("SingletonLock")))
+        self.assertFalse(is_profile_lock_error(RuntimeError("timeout waiting")))
+
+    def test_force_quit_rejects_short_path(self) -> None:
+        self.assertEqual(force_quit_profile_browsers("C:\\a"), 0)
+
+    def test_clear_stale_singleton_when_no_pids(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            profile = Path(tmp) / "account-manager" / "profiles" / "acc-1"
+            profile.mkdir(parents=True)
+            lock = profile / "SingletonLock"
+            cookie = profile / "SingletonCookie"
+            lock.write_text("deadhost-999999", encoding="utf-8")
+            cookie.write_text("x", encoding="utf-8")
+            with mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
+                return_value=[],
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock._pid_alive",
+                return_value=False,
+            ):
+                removed = clear_stale_singleton_locks(str(profile))
+            self.assertTrue(removed)
+            self.assertFalse(lock.exists())
+            self.assertFalse(cookie.exists())
+
+    def test_clear_stale_skips_when_pids_present(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            profile = Path(tmp) / "profiles" / "acc-2"
+            profile.mkdir(parents=True)
+            lock = profile / "SingletonLock"
+            lock.write_text("host-1", encoding="utf-8")
+            with mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
+                return_value=[1234],
+            ):
+                removed = clear_stale_singleton_locks(str(profile))
+            self.assertFalse(removed)
+            self.assertTrue(lock.exists())
+
+    def test_ensure_profile_free_quits_then_ok(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            profile = Path(tmp) / "profiles" / "acc-long-enough"
+            profile.mkdir(parents=True)
+            with mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
+                side_effect=[[111], []],
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.force_quit_profile_browsers",
+                return_value=1,
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.clear_stale_singleton_locks",
+                return_value=False,
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.time.sleep",
+            ):
+                result = ensure_profile_free(
+                    str(profile), wait_sec=0, raise_if_busy=True
+                )
+            self.assertTrue(result.freed)
+            self.assertEqual(result.pids_before, (111,))
+            self.assertEqual(result.pids_after, ())
+            self.assertEqual(result.killed_ops, 1)
+
+    def test_ensure_profile_free_raises_when_still_busy(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            profile = Path(tmp) / "profiles" / "acc-still-busy"
+            profile.mkdir(parents=True)
+            with mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
+                return_value=[222],
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.force_quit_profile_browsers",
+                return_value=1,
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.clear_stale_singleton_locks",
+                return_value=False,
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.time.sleep",
+            ):
+                with self.assertRaises(RuntimeError) as ctx:
+                    ensure_profile_free(str(profile), wait_sec=0, raise_if_busy=True)
+            self.assertIn(ERR_PROFILE_IN_USE, str(ctx.exception))
+
+    def test_ensure_profile_free_no_raise_when_requested(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            profile = Path(tmp) / "profiles" / "acc-keep-busy"
+            profile.mkdir(parents=True)
+            with mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.list_profile_browser_pids",
+                return_value=[333],
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.force_quit_profile_browsers",
+                return_value=0,
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.clear_stale_singleton_locks",
+                return_value=False,
+            ), mock.patch(
+                "jiangchang_skill_core.rpa.profile_lock.time.sleep",
+            ):
+                result = ensure_profile_free(
+                    str(profile), wait_sec=0, raise_if_busy=False
+                )
+            self.assertFalse(result.freed)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_unified_logging.py

@@ -45,6 +45,25 @@ class TestUnifiedLogging(unittest.TestCase):
         self.assertFalse(any(isinstance(h, TimedRotatingFileHandler) for h in handlers))
         self.assertTrue(any(isinstance(h, logging.NullHandler) for h in handlers))
 
+    def test_log_timestamp_uses_space_not_iso_t(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "skill.log")
+            logger_name = "test.logger.datefmt"
+            with patch.object(ul, "get_skill_log_file_path", return_value=log_path):
+                ul.setup_skill_logging("receive-order", logger_name)
+                ul.get_skill_logger().info("hello-datefmt")
+                ul._clear_logger_handlers(logging.getLogger(logger_name))
+            with open(log_path, encoding="utf-8") as f:
+                text = f.read()
+            self.assertRegex(
+                text,
+                r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \|",
+            )
+            self.assertNotRegex(
+                text,
+                r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2} \|",
+            )
+
     def test_attach_unified_file_handler_unwritable(self) -> None:
         logger_name = "test.logger.attach.unwritable"
         with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):