Merge commit 'ffda3f8738490e2128982bef7972a52d6707fc37'

# Conflicts:
#	web/src/components/layout/AppHeader.tsx
#	web/src/components/layout/MobileMenuSheet.tsx

Author: waterbang

src/cccc/cli/common.py

@@ -3,6 +3,7 @@
 import argparse
 import json
 import os
+import shutil
 import signal
 import socket
 import subprocess
@@ -124,6 +125,36 @@ def _cleanup_daemon_state_files(home: Path) -> None:
         path.unlink(missing_ok=True)
 
 
+def _same_home_daemon_lock_holder_pids(home: Path) -> list[int]:
+    if os.name != "posix":
+        return []
+    if not shutil.which("lsof"):
+        return []
+    lock_path = home / "daemon" / "ccccd.lock"
+    try:
+        proc = subprocess.run(
+            ["lsof", "-t", str(lock_path)],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except Exception:
+        return []
+    if int(proc.returncode or 0) not in {0, 1}:
+        return []
+
+    pids: list[int] = []
+    for line in str(proc.stdout or "").splitlines():
+        text = str(line or "").strip()
+        if not text.isdigit():
+            continue
+        pid = int(text)
+        if pid <= 0 or pid == os.getpid():
+            continue
+        pids.append(pid)
+    return sorted(set(pids))
+
+
 def _same_home_daemon_pids(home: Path) -> list[int]:
     if os.name != "posix":
         return []
@@ -132,7 +163,7 @@ def _same_home_daemon_pids(home: Path) -> list[int]:
     default_home = str(ensure_home().resolve())
     proc_root = Path("/proc")
     if not proc_root.is_dir():
-        return []
+        return _same_home_daemon_lock_holder_pids(home)
     pids: list[int] = []
     try:
         proc_dirs = list(proc_root.iterdir())

src/cccc/contracts/v1/group_space.py

@@ -67,8 +67,10 @@ class SpaceJob(BaseModel):
     remote_space_id: str = ""
     kind: SpaceJobKind = "context_sync"
     payload: Dict[str, Any] = Field(default_factory=dict)
+    payload_ref: str = ""
     result: Dict[str, Any] = Field(default_factory=dict)
     payload_digest: str = ""
+    payload_bytes: int = 0
     idempotency_key: str = ""
     state: SpaceJobState = "pending"
     attempt: int = 0

src/cccc/contracts/v1/message.py

@@ -50,6 +50,9 @@ class ChatMessageData(BaseModel):
     source_user_name: Optional[str] = None  # External IM sender display name
     source_user_id: Optional[str] = None  # External IM sender platform user id
     mention_user_ids: Optional[List[str]] = None  # External IM real-mention targets
+    sender_title: Optional[str] = None  # Immutable sender title snapshot for message rendering
+    sender_runtime: Optional[str] = None  # Immutable sender runtime snapshot for message rendering
+    sender_avatar_path: Optional[str] = None  # Immutable blob-backed sender avatar path
 
     # Cross-group provenance (for relays/forwarding)
     src_group_id: Optional[str] = None
@@ -68,6 +71,7 @@ class ChatMessageData(BaseModel):
 
     # Streaming
     stream_id: Optional[str] = None  # Links final message to its stream
+    pending_event_id: Optional[str] = None  # Stable turn-scoped id to reconcile transient UI bubbles
 
     # Metadata
     client_id: Optional[str] = None  # Client-generated idempotency key

src/cccc/daemon/actor_runtime_cache.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import threading
+from typing import Any, Dict
+
+_LOCK = threading.Lock()
+_BY_GROUP: Dict[str, Dict[str, Dict[str, Any]]] = {}
+
+
+def replace_group_runtime(group_id: str, actors: Dict[str, Dict[str, Any]]) -> None:
+    gid = str(group_id or "").strip()
+    if not gid:
+        return
+    normalized: Dict[str, Dict[str, Any]] = {}
+    for actor_id, payload in actors.items():
+        aid = str(actor_id or "").strip()
+        if not aid or not isinstance(payload, dict):
+            continue
+        normalized[aid] = {
+            "running": bool(payload.get("running")),
+            "runner_effective": str(payload.get("runner_effective") or "").strip() or "pty",
+            "idle_seconds": payload.get("idle_seconds"),
+            "effective_working_state": str(payload.get("effective_working_state") or "").strip() or "stopped",
+            "effective_working_reason": str(payload.get("effective_working_reason") or "").strip(),
+            "effective_working_updated_at": payload.get("effective_working_updated_at"),
+            "effective_active_task_id": payload.get("effective_active_task_id"),
+        }
+    with _LOCK:
+        _BY_GROUP[gid] = normalized
+
+
+def get_group_runtime(group_id: str) -> Dict[str, Dict[str, Any]]:
+    gid = str(group_id or "").strip()
+    if not gid:
+        return {}
+    with _LOCK:
+        current = _BY_GROUP.get(gid) or {}
+        return {actor_id: dict(payload) for actor_id, payload in current.items()}

src/cccc/daemon/actors/actor_lifecycle_ops.py

@@ -11,6 +11,8 @@
 from ...kernel.group import load_group
 from ...kernel.ledger import append_event
 from ...kernel.permissions import require_actor_permission
+from ...kernel.runtime import inject_runtime_home_env
+from ..codex_app_sessions import SUPERVISOR as codex_app_supervisor
 from ...runners import headless as headless_runner
 from ...runners import pty as pty_runner
 from ...util.conv import coerce_bool
@@ -114,7 +116,12 @@ def handle_actor_stop(
         actor = update_actor(group, actor_id, {"enabled": False})
         runner_kind = str(actor.get("runner") or "pty").strip()
         runner_effective = effective_runner_kind(runner_kind)
-        if runner_effective == "headless":
+        runtime = str(actor.get("runtime") or "codex").strip() or "codex"
+        if runtime == "codex" and runner_effective == "headless":
+            codex_app_supervisor.stop_actor(group_id=group.group_id, actor_id=actor_id)
+            remove_headless_state(group.group_id, actor_id)
+            remove_pty_state_if_pid(group.group_id, actor_id, pid=0)
+        elif runner_effective == "headless":
             headless_runner.SUPERVISOR.stop_actor(group_id=group.group_id, actor_id=actor_id)
             remove_headless_state(group.group_id, actor_id)
             remove_pty_state_if_pid(group.group_id, actor_id, pid=0)
@@ -211,8 +218,12 @@ def handle_actor_restart(
             is_admin=is_admin,
         )
         runner_kind = str(actor.get("runner") or "pty").strip()
-        runner_effective = effective_runner_kind(runner_kind)
-        if runner_effective == "headless":
+        runtime = str(actor.get("runtime") or "codex").strip() or "codex"
+        if runtime == "codex" and effective_runner_kind(runner_kind) == "headless":
+            codex_app_supervisor.stop_actor(group_id=group.group_id, actor_id=actor_id)
+            remove_headless_state(group.group_id, actor_id)
+            remove_pty_state_if_pid(group.group_id, actor_id, pid=0)
+        elif effective_runner_kind(runner_kind) == "headless":
             headless_runner.SUPERVISOR.stop_actor(group_id=group.group_id, actor_id=actor_id)
             remove_headless_state(group.group_id, actor_id)
             remove_pty_state_if_pid(group.group_id, actor_id, pid=0)
@@ -308,6 +319,12 @@ def handle_actor_restart(
         runner_kind = str(launch_spec["runner"])
         runner_effective = str(launch_spec["effective_runner"])
         runtime = str(launch_spec["runtime"])
+        effective_env = inject_runtime_home_env(
+            launch_spec["merged_env"],
+            runtime=runtime,
+            group_id=group.group_id,
+            actor_id=actor_id,
+        )
         if runner_effective != "headless":
             try:
                 mcp_ready = bool(ensure_mcp_installed(runtime, cwd))
@@ -316,12 +333,19 @@ def handle_actor_restart(
             if not mcp_ready:
                 return _error("actor_restart_failed", f"failed to install MCP for runtime: {runtime}")
 
-        if runner_effective == "headless":
+        if runtime == "codex" and runner_effective == "headless":
+            codex_app_supervisor.start_actor(
+                group_id=group.group_id,
+                actor_id=actor_id,
+                cwd=cwd,
+                env=dict(inject_actor_context_env(effective_env, group_id=group.group_id, actor_id=actor_id)),
+            )
+        elif runner_effective == "headless":
             headless_runner.SUPERVISOR.start_actor(
                 group_id=group.group_id,
                 actor_id=actor_id,
                 cwd=cwd,
-                env=dict(inject_actor_context_env(launch_spec["merged_env"], group_id=group.group_id, actor_id=actor_id)),
+                env=dict(inject_actor_context_env(effective_env, group_id=group.group_id, actor_id=actor_id)),
             )
             try:
                 write_headless_state(group.group_id, actor_id)
@@ -333,9 +357,8 @@ def handle_actor_restart(
                 actor_id=actor_id,
                 cwd=cwd,
                 command=launch_spec["effective_command"],
-                env=prepare_pty_env(
-                    inject_actor_context_env(launch_spec["merged_env"], group_id=group.group_id, actor_id=actor_id)
-                ),
+                env=prepare_pty_env(inject_actor_context_env(effective_env, group_id=group.group_id, actor_id=actor_id)),
+                runtime=runtime,
                 max_backlog_bytes=pty_backlog_bytes(),
             )
             try:
@@ -346,6 +369,13 @@ def handle_actor_restart(
             ContextStorage(group).clear_agent_status_if_present(actor_id)
         except Exception:
             pass
+        try:
+            if str(group.doc.get("state") or "").strip() == "stopped":
+                group.doc["state"] = "active"
+            group.doc["running"] = True
+            group.save()
+        except Exception:
+            pass
 
     maybe_reset_automation_on_foreman_change(group, before_foreman_id=before_foreman)
     event = append_event(
@@ -354,7 +384,11 @@ def handle_actor_restart(
         group_id=group.group_id,
         scope_key="",
         by=by,
-        data={"actor_id": actor_id, "runner": str(actor.get("runner") or "pty")},
+        data={
+            "actor_id": actor_id,
+            "runner": str(actor.get("runner") or "pty"),
+            "runner_effective": runner_effective,
+        },
     )
 
     from ...kernel.events import publish_event

src/cccc/daemon/actors/actor_membership_ops.py

@@ -7,12 +7,14 @@
 from ...contracts.v1 import DaemonError, DaemonResponse
 from ...kernel.actors import find_actor, list_actors, remove_actor
 from ...kernel.context import ContextStorage
+from ...kernel.events import publish_event
 from ...kernel.group import load_group
 from ...kernel.ledger import append_event
 from ...kernel.permissions import require_actor_permission
 from ...runners import headless as headless_runner
 from ...runners import pty as pty_runner
 from ...util.conv import coerce_bool
+from ..codex_app_sessions import SUPERVISOR as codex_app_supervisor
 from ..context.context_ops import _schedule_summary_snapshot_rebuild
 
 
@@ -47,6 +49,7 @@ def handle_actor_remove(
         if isinstance(actor_doc, dict):
             avatar_rel_path = str(actor_doc.get("avatar_asset_path") or "").strip()
         remove_actor(group, actor_id)
+        codex_app_supervisor.stop_actor(group_id=group.group_id, actor_id=actor_id)
         pty_runner.SUPERVISOR.stop_actor(group_id=group.group_id, actor_id=actor_id)
         remove_pty_state_if_pid(group.group_id, actor_id, pid=0)
         headless_runner.SUPERVISOR.stop_actor(group_id=group.group_id, actor_id=actor_id)
@@ -84,6 +87,7 @@ def handle_actor_remove(
         by=by,
         data={"actor_id": actor_id},
     )
+    publish_event("actor.remove", {"group_id": group.group_id, "actor_id": actor_id})
     maybe_reset_automation_on_foreman_change(group, before_foreman_id=before_foreman)
     return DaemonResponse(ok=True, result={"actor_id": actor_id, "event": event})
 

src/cccc/daemon/actors/actor_ops.py

@@ -5,15 +5,17 @@
 from typing import Any, Callable, Dict, Optional
 
 from ...contracts.v1 import DaemonError, DaemonResponse
+from ..actor_runtime_cache import get_group_runtime
+from ..codex_app_sessions import SUPERVISOR as codex_app_supervisor
 from ...kernel.actors import find_actor, get_effective_role, list_actors
 from ...kernel.context import ContextStorage
 from ...kernel.group import load_group
 from ...kernel.query_projections import get_actor_list_projection
 from ...kernel.working_state import DEFAULT_PTY_TERMINAL_SIGNAL_TAIL_BYTES, derive_effective_working_state
-from ..context.context_ops import _agent_state_to_dict
-from .private_env_ops import mask_private_env_value
 from ...runners import headless as headless_runner
 from ...runners import pty as pty_runner
+from ..context.context_ops import _agent_state_to_dict
+from .private_env_ops import mask_private_env_value
 from ...util.conv import coerce_bool
 
 
@@ -47,6 +49,7 @@ def handle_actor_list(
             actors.append(item)
     else:
         actors = get_actor_list_projection(group)
+    runtime_snapshot = get_group_runtime(group_id)
     storage = ContextStorage(group)
     agent_rows = [_agent_state_to_dict(agent) for agent in storage.load_agents().agents]
     agent_state_by_id = {
@@ -59,39 +62,61 @@ def handle_actor_list(
         if not aid:
             continue
         runner_kind = str(actor.get("runner") or "pty").strip()
+        snap = runtime_snapshot.get(aid) if isinstance(runtime_snapshot.get(aid), dict) else {}
+        if snap:
+            actor["running"] = bool(snap.get("running"))
+            actor["idle_seconds"] = snap.get("idle_seconds")
+            snap_runner_effective = str(snap.get("runner_effective") or runner_kind or "pty")
+            if snap_runner_effective == "headless" or snap_runner_effective != runner_kind:
+                actor["runner_effective"] = snap_runner_effective
+            else:
+                actor.pop("runner_effective", None)
+            for key in (
+                "effective_working_state",
+                "effective_working_reason",
+                "effective_working_updated_at",
+                "effective_active_task_id",
+            ):
+                if key in snap:
+                    actor[key] = snap.get(key)
+            continue
         effective_runner = effective_runner_kind(runner_kind)
-        idle_seconds = None
+        runtime = str(actor.get("runtime") or "").strip()
         headless_state = None
-        if effective_runner == "headless":
-            actor["running"] = headless_runner.SUPERVISOR.actor_running(group_id, aid)
+        running = False
+        idle_seconds = None
+        pty_terminal_text = ""
+        if runtime.lower() == "codex" and effective_runner == "headless":
+            state = codex_app_supervisor.get_state(group_id=group_id, actor_id=aid)
+            headless_state = state.model_dump() if hasattr(state, "model_dump") else (dict(state) if isinstance(state, dict) else None)
+            running = bool(state is not None and codex_app_supervisor.actor_running(group_id, aid))
+        elif effective_runner == "headless":
             state = headless_runner.SUPERVISOR.get_state(group_id=group_id, actor_id=aid)
-            headless_state = state.model_dump() if state is not None else None
-            actor["idle_seconds"] = None
+            headless_state = state.model_dump() if hasattr(state, "model_dump") else (dict(state) if isinstance(state, dict) else None)
+            running = bool(state is not None and headless_runner.SUPERVISOR.actor_running(group_id, aid))
         else:
-            actor["running"] = pty_runner.SUPERVISOR.actor_running(group_id, aid)
-            idle_seconds = (
-                pty_runner.SUPERVISOR.idle_seconds(group_id=group_id, actor_id=aid)
-                if actor["running"]
-                else None
-            )
-            actor["idle_seconds"] = idle_seconds
-        pty_terminal_text = ""
-        if effective_runner == "pty" and actor["running"]:
-            try:
-                pty_terminal_text = pty_runner.SUPERVISOR.tail_output(
-                    group_id=group_id,
-                    actor_id=aid,
-                    max_bytes=DEFAULT_PTY_TERMINAL_SIGNAL_TAIL_BYTES,
-                ).decode("utf-8", errors="replace")
-            except Exception:
-                pty_terminal_text = ""
-        if effective_runner != runner_kind:
+            running = bool(pty_runner.SUPERVISOR.actor_running(group_id, aid))
+            idle_seconds = pty_runner.SUPERVISOR.idle_seconds(group_id=group_id, actor_id=aid) if running else None
+            if running:
+                try:
+                    pty_terminal_text = pty_runner.SUPERVISOR.tail_output(
+                        group_id=group_id,
+                        actor_id=aid,
+                        max_bytes=DEFAULT_PTY_TERMINAL_SIGNAL_TAIL_BYTES,
+                    ).decode("utf-8", errors="replace")
+                except Exception:
+                    pty_terminal_text = ""
+        actor["running"] = running
+        actor["idle_seconds"] = idle_seconds
+        if effective_runner == "headless" or effective_runner != runner_kind:
             actor["runner_effective"] = effective_runner
+        else:
+            actor.pop("runner_effective", None)
         actor.update(
             derive_effective_working_state(
-                running=bool(actor.get("running")),
+                running=running,
                 effective_runner=effective_runner,
-                runtime=str(actor.get("runtime") or ""),
+                runtime=runtime,
                 idle_seconds=idle_seconds,
                 pty_terminal_text=pty_terminal_text,
                 agent_state=agent_state_by_id.get(aid),