Stabilize runtime dock and headless streaming

- move headless live output out of chat bubbles into dock ticker/inspector
- align runtime dock labels, ticker cache, and hidden-tab stream recovery
- persist actor.activity state transitions and fix WeCom media decrypt portability

Author: ChesterRa

pyproject.toml

@@ -14,6 +14,7 @@ authors = [
 license = "Apache-2.0"
 keywords = ["orchestrator","ai","rfd","pair","collaboration"]
 dependencies = [
+  "cryptography>=41.0.0,<47.0",
   "PyYAML>=6.0,<7.0",
   "dingtalk-stream>=0.24.3",
   "fastapi>=0.110,<1.0",

src/cccc/contracts/v1/event.py

@@ -30,6 +30,7 @@
     "actor.stop",
     "actor.restart",
     "actor.remove",
+    "actor.activity",
     "context.sync",
     "chat.message",
     "chat.stream",
@@ -167,6 +168,13 @@ class ActorLifecycleData(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
 
+class ActorActivityData(BaseModel):
+    """Periodic runtime status snapshot for running actors."""
+    actors: List[Dict[str, Any]] = Field(default_factory=list)
+
+    model_config = ConfigDict(extra="allow")
+
+
 class ContextSyncData(BaseModel):
     version: str = ""
     changes: List[Dict[str, Any]] = Field(default_factory=list)
@@ -242,6 +250,7 @@ class Event(BaseModel):
     "actor.stop": ActorLifecycleData,
     "actor.restart": ActorLifecycleData,
     "actor.remove": ActorLifecycleData,
+    "actor.activity": ActorActivityData,
     "context.sync": ContextSyncData,
     "chat.message": ChatMessageData,
     "chat.stream": ChatStreamData,

src/cccc/daemon/serve_ops.py

@@ -293,6 +293,7 @@ def start_actor_activity_thread(
 
     def _actor_activity_loop() -> None:
         interval = max(1.0, float(interval_seconds or 1.0))
+        prev_runtime_by_group: Dict[str, Dict[str, Dict[str, Any]]] = {}
         while not stop_event.is_set():
             try:
                 groups_base = home / "groups"
@@ -377,15 +378,72 @@ def _actor_activity_loop() -> None:
                             actors_data.append(payload)
                             actors_snapshot[aid] = payload
                         replace_group_runtime(gid, actors_snapshot)
+
+                        # Broadcast running actors in-memory (daemon IPC).
                         if actors_data:
-                            event_broadcaster.publish({
+                            activity_event = {
                                 "id": uuid.uuid4().hex,
                                 "kind": "actor.activity",
                                 "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                                 "group_id": gid,
                                 "by": "system",
                                 "data": {"actors": actors_data},
-                            })
+                            }
+                            event_broadcaster.publish(activity_event)
+
+                        # Detect working-state transitions so the web port
+                        # (which tails the ledger file) also receives the
+                        # update.  Without this, actor.activity only flows
+                        # through the in-memory EventBroadcaster and never
+                        # reaches the web frontend.
+                        prev_snapshot = prev_runtime_by_group.get(gid, {})
+                        state_changed = len(actors_snapshot) != len(prev_snapshot)
+                        if not state_changed:
+                            for aid, payload in actors_snapshot.items():
+                                prev_actor = prev_snapshot.get(aid)
+                                if prev_actor is None or payload.get("effective_working_state") != prev_actor.get("effective_working_state"):
+                                    state_changed = True
+                                    break
+                        # Emit "stopped" entries for actors that disappeared
+                        # (crashed/stopped since last tick) so the web
+                        # frontend can clear stale "working" halos.
+                        stopped_entries: list[Dict[str, Any]] = []
+                        for prev_aid, prev_actor in prev_snapshot.items():
+                            if prev_aid not in actors_snapshot:
+                                prev_runner = str(prev_actor.get("runner_effective") or "pty").strip() or "pty"
+                                state_changed = True
+                                stopped_entries.append({
+                                    "id": prev_aid,
+                                    "running": False,
+                                    "runner_effective": prev_runner,
+                                    "idle_seconds": None,
+                                    "effective_working_state": "stopped",
+                                    "effective_working_reason": "runner_not_running",
+                                    "effective_working_updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+                                    "effective_active_task_id": None,
+                                })
+                        prev_runtime_by_group[gid] = {
+                            aid: {
+                                "effective_working_state": p.get("effective_working_state"),
+                                "runner_effective": p.get("runner_effective"),
+                            }
+                            for aid, p in actors_snapshot.items()
+                        }
+                        if state_changed:
+                            ledger_actors = actors_data + stopped_entries
+                            if ledger_actors:
+                                try:
+                                    from ..kernel.ledger import append_event
+                                    append_event(
+                                        group.ledger_path,
+                                        kind="actor.activity",
+                                        group_id=gid,
+                                        scope_key="",
+                                        by="system",
+                                        data={"actors": ledger_actors},
+                                    )
+                                except Exception:
+                                    pass
             except Exception as e:
                 _log_loop_error("actor_activity_tick failed", e)
             stop_event.wait(interval)

src/cccc/ports/im/adapters/wecom.py

@@ -18,7 +18,6 @@
 import mimetypes
 import os
 import random
-import subprocess
 import threading
 import time
 import urllib.parse
@@ -28,6 +27,8 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
 from .base import IMAdapter, OutboundStreamHandle
 
 # WeCom API limits
@@ -47,7 +48,6 @@
 # Keep the latest callback handle per chat for the lifetime of this bridge
 # process. We only need a bounded cache, not a time-based expiry.
 REPLY_REF_MAX_ENTRIES = 256
-WECOM_MEDIA_CIPHER = "aes-256-cbc"
 WECOM_MEDIA_BLOCK_SIZE = 32
 
 
@@ -249,30 +249,12 @@ def _strip_pkcs7_padding(self, raw: bytes, block_size: int = WECOM_MEDIA_BLOCK_S
     def _decrypt_media_bytes(self, encrypted: bytes, aes_key: str) -> bytes:
         key = self._decode_media_aes_key(aes_key)
         iv = key[:16]
-        cmd = [
-            "openssl",
-            "enc",
-            "-d",
-            f"-{WECOM_MEDIA_CIPHER}",
-            "-nopad",
-            "-K",
-            key.hex(),
-            "-iv",
-            iv.hex(),
-        ]
         try:
-            proc = subprocess.run(
-                cmd,
-                input=encrypted,
-                capture_output=True,
-                check=False,
-            )
+            decryptor = Cipher(algorithms.AES(key), modes.CBC(iv)).decryptor()
+            decrypted = decryptor.update(encrypted) + decryptor.finalize()
         except Exception as e:
-            raise ValueError(f"wecom media decrypt command failed: {e}") from e
-        if proc.returncode != 0:
-            stderr = proc.stderr.decode("utf-8", errors="replace").strip()
-            raise ValueError(f"wecom media decrypt failed: {stderr or f'openssl exit {proc.returncode}'}")
-        return self._strip_pkcs7_padding(proc.stdout)
+            raise ValueError(f"wecom media decrypt failed: {e}") from e
+        return self._strip_pkcs7_padding(decrypted)
 
     def _upload_media(self, raw: bytes, filename: str, media_type: str) -> str:
         boundary = "----cccc" + uuid.uuid4().hex

tests/test_codex_app_flow.py

@@ -1539,6 +1539,168 @@ def publish(self, event: dict) -> None:
         finally:
             cleanup()
 
+    def test_actor_activity_thread_writes_ledger_on_state_change(self) -> None:
+        """actor.activity should be written to ledger on working-state transitions."""
+        from cccc.daemon.serve_ops import start_actor_activity_thread
+        from cccc.kernel.actors import add_actor
+        from cccc.kernel.group import create_group, load_group
+        from cccc.kernel.registry import load_registry
+
+        home, cleanup = self._with_home()
+        try:
+            reg = load_registry()
+            created = create_group(reg, title="codex-ledger-activity", topic="")
+            group = load_group(created.group_id)
+            self.assertIsNotNone(group)
+            add_actor(group, actor_id="peer1", title="Peer 1", runtime="codex", runner="headless")  # type: ignore[arg-type]
+            group.save()  # type: ignore[union-attr]
+
+            # Re-load to get fresh ledger_path
+            group = load_group(created.group_id)
+            self.assertIsNotNone(group)
+            ledger_path = group.ledger_path  # type: ignore[union-attr]
+
+            class _Broadcaster:
+                def publish(self, event: dict) -> None:
+                    pass
+
+            status_holder = {"status": "working", "running": True}
+
+            class _CodexSupervisor:
+                @staticmethod
+                def get_state(group_id: str, actor_id: str) -> dict:
+                    return {
+                        "group_id": group_id,
+                        "actor_id": actor_id,
+                        "status": status_holder["status"],
+                        "current_task_id": "turn-1",
+                        "updated_at": "2026-04-02T10:00:00Z",
+                    }
+
+                @staticmethod
+                def actor_running(_group_id: str, _actor_id: str) -> bool:
+                    return bool(status_holder.get("running", True))
+
+            stop_event = threading.Event()
+            thread = start_actor_activity_thread(
+                stop_event=stop_event,
+                home=Path(home),
+                pty_supervisor=object(),
+                headless_supervisor=object(),
+                codex_supervisor=_CodexSupervisor(),
+                event_broadcaster=_Broadcaster(),
+                load_group=load_group,
+                interval_seconds=1.0,
+            )
+            try:
+                # First tick runs immediately: new actor → state_changed → writes to ledger
+                time.sleep(0.25)
+                # Verify ledger has actor.activity
+                import json
+                lines = ledger_path.read_text(encoding="utf-8").strip().split("\n")
+                activity_lines = [json.loads(line) for line in lines if '"actor.activity"' in line]
+                self.assertTrue(activity_lines, "First tick should write actor.activity to ledger")
+                self.assertEqual(activity_lines[-1]["data"]["actors"][0]["effective_working_state"], "working")
+
+                initial_count = len(activity_lines)
+                # Wait another tick (>1s interval) — no state change → no new ledger write
+                time.sleep(1.3)
+                lines2 = ledger_path.read_text(encoding="utf-8").strip().split("\n")
+                activity_lines2 = [json.loads(line) for line in lines2 if '"actor.activity"' in line]
+                self.assertEqual(len(activity_lines2), initial_count, "No state change should not add ledger entries")
+
+                # Change state: working → idle → should write to ledger on next tick
+                status_holder["status"] = "idle"
+                time.sleep(1.3)
+                lines3 = ledger_path.read_text(encoding="utf-8").strip().split("\n")
+                activity_lines3 = [json.loads(line) for line in lines3 if '"actor.activity"' in line]
+                self.assertGreater(len(activity_lines3), initial_count, "State change should add ledger entry")
+                self.assertEqual(activity_lines3[-1]["data"]["actors"][0]["effective_working_state"], "idle")
+
+                # Simulate actor stopping (actor_running returns False)
+                idle_count = len(activity_lines3)
+                status_holder["running"] = False
+                time.sleep(1.3)
+                lines4 = ledger_path.read_text(encoding="utf-8").strip().split("\n")
+                activity_lines4 = [json.loads(line) for line in lines4 if '"actor.activity"' in line]
+                self.assertGreater(len(activity_lines4), idle_count, "Actor stop should add ledger entry")
+                last_event = activity_lines4[-1]
+                stopped_actors = [a for a in last_event["data"]["actors"] if a["id"] == "peer1"]
+                self.assertEqual(len(stopped_actors), 1, "Stopped actor should appear in event")
+                self.assertEqual(stopped_actors[0]["effective_working_state"], "stopped")
+                self.assertFalse(stopped_actors[0]["running"])
+            finally:
+                stop_event.set()
+                thread.join(timeout=1.0)
+        finally:
+            cleanup()
+
+    def test_actor_activity_thread_preserves_runner_on_stopped_entry(self) -> None:
+        from cccc.daemon.serve_ops import start_actor_activity_thread
+        from cccc.kernel.actors import add_actor
+        from cccc.kernel.group import create_group, load_group
+        from cccc.kernel.registry import load_registry
+
+        home, cleanup = self._with_home()
+        try:
+            reg = load_registry()
+            created = create_group(reg, title="pty-ledger-activity", topic="")
+            group = load_group(created.group_id)
+            self.assertIsNotNone(group)
+            add_actor(group, actor_id="peer1", title="Peer 1", runtime="codex", runner="pty")  # type: ignore[arg-type]
+            group.save()  # type: ignore[union-attr]
+
+            group = load_group(created.group_id)
+            self.assertIsNotNone(group)
+            ledger_path = group.ledger_path  # type: ignore[union-attr]
+            status_holder = {"running": True}
+
+            class _PtySupervisor:
+                @staticmethod
+                def actor_running(_group_id: str, _actor_id: str) -> bool:
+                    return bool(status_holder.get("running", True))
+
+                @staticmethod
+                def idle_seconds(*, group_id: str, actor_id: str) -> float:
+                    return 0.0
+
+                @staticmethod
+                def terminal_override(*, group_id: str, actor_id: str):
+                    return None
+
+            class _Broadcaster:
+                def publish(self, event: dict) -> None:
+                    pass
+
+            stop_event = threading.Event()
+            thread = start_actor_activity_thread(
+                stop_event=stop_event,
+                home=Path(home),
+                pty_supervisor=_PtySupervisor(),
+                headless_supervisor=object(),
+                codex_supervisor=object(),
+                event_broadcaster=_Broadcaster(),
+                load_group=load_group,
+                interval_seconds=1.0,
+            )
+            try:
+                time.sleep(0.25)
+                status_holder["running"] = False
+                time.sleep(1.3)
+
+                lines = ledger_path.read_text(encoding="utf-8").strip().split("\n")
+                activity_lines = [json.loads(line) for line in lines if '"actor.activity"' in line]
+                self.assertTrue(activity_lines, "Actor stop should write actor.activity to ledger")
+                stopped_actors = [a for a in activity_lines[-1]["data"]["actors"] if a["id"] == "peer1"]
+                self.assertEqual(len(stopped_actors), 1)
+                self.assertEqual(stopped_actors[0]["effective_working_state"], "stopped")
+                self.assertEqual(stopped_actors[0]["runner_effective"], "pty")
+            finally:
+                stop_event.set()
+                thread.join(timeout=1.0)
+        finally:
+            cleanup()
+
     def test_codex_session_persists_headless_state_file(self) -> None:
         from cccc.daemon.codex_app_sessions import CodexAppSession
         from cccc.daemon.runner_state_ops import headless_state_path

tests/test_wecom_adapter_core.py

@@ -1,5 +1,6 @@
 """Tests for WecomAdapter core functionality (Steps 7-13)."""
 
+import subprocess
 import sys
 import tempfile
 import threading
@@ -8,6 +9,16 @@
 from pathlib import Path
 from unittest.mock import patch
 
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
+
+def _encrypt_wecom_media_for_test(raw: bytes, aes_key: str) -> bytes:
+    key = aes_key.encode("utf-8")
+    pad = 32 - (len(raw) % 32)
+    padded = raw + bytes([pad]) * pad
+    encryptor = Cipher(algorithms.AES(key), modes.CBC(key[:16])).encryptor()
+    return encryptor.update(padded) + encryptor.finalize()
+
 
 class TestWecomAuthFrames(unittest.TestCase):
     """WeCom AI Bot auth now uses bot_id + secret over WebSocket."""
@@ -736,6 +747,25 @@ def fake_urlopen(req, timeout=0):
         self.assertEqual(raw, b"plain-bytes")
         mock_decrypt.assert_called_once_with(b"encrypted-bytes", "12345678901234567890123456789012")
 
+    def test_download_attachment_decrypts_direct_url_without_external_openssl(self):
+        adapter = self._make_adapter()
+        aes_key = "12345678901234567890123456789012"
+        encrypted = _encrypt_wecom_media_for_test(b"plain attachment bytes", aes_key)
+
+        def fake_urlopen(req, timeout=0):
+            self.assertEqual(req.full_url, "https://example.test/media.enc")
+            self.assertEqual(timeout, 60)
+            return _FakeHttpResponse(encrypted)
+
+        with patch("cccc.ports.im.adapters.wecom.urllib.request.urlopen", side_effect=fake_urlopen):
+            with patch.object(subprocess, "run", side_effect=AssertionError("external openssl should not be used")):
+                raw = adapter.download_attachment({
+                    "download_url": "https://example.test/media.enc",
+                    "aeskey": aes_key,
+                })
+
+        self.assertEqual(raw, b"plain attachment bytes")
+
     def test_download_attachment_uses_direct_url_without_decrypt_when_no_aeskey(self):
         adapter = self._make_adapter()