Fix headless reply parity and runtime dock polish

Author: ChesterRa

src/cccc/daemon/claude_app_sessions.py

@@ -8,7 +8,7 @@
 from pathlib import Path
 from typing import Any, Callable, Dict, Optional
 
-from ...contracts.v1 import ChatMessageData, ChatStreamData, DaemonError, DaemonResponse
+from ...contracts.v1 import ChatMessageData, ChatStreamData, DaemonError, DaemonResponse, SystemNotifyData
 from ...kernel.actors import find_actor, list_actors, resolve_recipient_tokens
 from ...kernel.group import get_group_state, load_group, set_group_state
 from ...kernel.inbox import find_event_with_chat_ack, is_message_for_actor
@@ -26,9 +26,11 @@
 from ..claude_app_sessions import SUPERVISOR as claude_app_supervisor
 from ..codex_app_sessions import SUPERVISOR as codex_app_supervisor
 from .delivery import (
+    append_mcp_reply_reminder,
+    emit_system_notify,
+    should_auto_mark_on_delivery,
     flush_pending_messages,
     get_headless_targets_for_message,
-    maybe_auto_mark_delivered_event,
     queue_chat_message,
     request_flush_pending_messages,
 )
@@ -270,21 +272,18 @@ def _notify_headless_targets(
         for actor_id in headless_targets:
             if actor_id in skip_ids:
                 continue
-            append_event(
-                group.ledger_path,
-                kind="system.notify",
-                group_id=group.group_id,
-                scope_key="",
+            emit_system_notify(
+                group,
                 by="system",
-                data={
-                    "kind": "info",
-                    "priority": notify_priority,
-                    "title": notify_title,
-                    "message": f"New message from {by}. Check your inbox.",
-                    "target_actor_id": actor_id,
-                    "requires_ack": False,
-                    "context": {"event_id": event_id, "from": by},
-                },
+                notify=SystemNotifyData(
+                    kind="info",
+                    priority=notify_priority,
+                    title=notify_title,
+                    message=f"New message from {by}. Check your inbox.",
+                    target_actor_id=actor_id,
+                    requires_ack=False,
+                    context={"event_id": event_id, "from": by},
+                ),
             )
     except Exception:
         pass
@@ -467,7 +466,9 @@ def handle_send(
         src_group_id=src_group_id,
         src_event_id=src_event_id,
     )
+    headless_delivery_text = append_mcp_reply_reminder(delivery_text)
     actors = list_actors(group)
+    auto_mark_on_delivery = should_auto_mark_on_delivery(group)
     skip_headless_notify_actor_ids: set[str] = set()
     logger.debug(f"[SEND] group={group_id} text={text[:30]!r} actors={[a.get('id') for a in actors]} effective_to={effective_to}")
     for actor in actors:
@@ -493,11 +494,12 @@ def handle_send(
             delivered = bool(codex_app_supervisor.submit_user_message(
                 group_id=group.group_id,
                 actor_id=actor_id,
-                text=delivery_text,
+                text=headless_delivery_text,
                 event_id=event_id,
+                ts=event_ts,
                 attachments=attachments,
             ))
-            if delivered and maybe_auto_mark_delivered_event(group, actor_id=actor_id, event_id=event_id, ts=event_ts):
+            if delivered and auto_mark_on_delivery:
                 skip_headless_notify_actor_ids.add(actor_id)
         elif (
             runtime == "claude"
@@ -507,11 +509,12 @@ def handle_send(
             delivered = bool(claude_app_supervisor.submit_user_message(
                 group_id=group.group_id,
                 actor_id=actor_id,
-                text=delivery_text,
+                text=headless_delivery_text,
                 event_id=event_id,
+                ts=event_ts,
                 attachments=attachments,
             ))
-            if delivered and maybe_auto_mark_delivered_event(group, actor_id=actor_id, event_id=event_id, ts=event_ts):
+            if delivered and auto_mark_on_delivery:
                 skip_headless_notify_actor_ids.add(actor_id)
         elif effective_runner_kind(runner_kind) == "pty":
             queue_chat_message(
@@ -725,6 +728,8 @@ def handle_reply(
         refs=refs,
         attachments=attachments,
     )
+    headless_delivery_text = append_mcp_reply_reminder(delivery_text)
+    auto_mark_on_delivery = should_auto_mark_on_delivery(group)
     skip_headless_notify_actor_ids: set[str] = set()
     for actor in list_actors(group):
         if not isinstance(actor, dict):
@@ -744,12 +749,13 @@ def handle_reply(
             delivered = bool(codex_app_supervisor.submit_user_message(
                 group_id=group.group_id,
                 actor_id=actor_id,
-                text=delivery_text,
+                text=headless_delivery_text,
                 event_id=event_id,
+                ts=event_ts,
                 reply_to=target_event_id or reply_to,
                 attachments=attachments,
             ))
-            if delivered and maybe_auto_mark_delivered_event(group, actor_id=actor_id, event_id=event_id, ts=event_ts):
+            if delivered and auto_mark_on_delivery:
                 skip_headless_notify_actor_ids.add(actor_id)
         elif (
             runtime == "claude"
@@ -759,12 +765,13 @@ def handle_reply(
             delivered = bool(claude_app_supervisor.submit_user_message(
                 group_id=group.group_id,
                 actor_id=actor_id,
-                text=delivery_text,
+                text=headless_delivery_text,
                 event_id=event_id,
+                ts=event_ts,
                 reply_to=target_event_id or reply_to,
                 attachments=attachments,
             ))
-            if delivered and maybe_auto_mark_delivered_event(group, actor_id=actor_id, event_id=event_id, ts=event_ts):
+            if delivered and auto_mark_on_delivery:
                 skip_headless_notify_actor_ids.add(actor_id)
         elif effective_runner_kind(runner_kind) == "pty":
             queue_chat_message(

src/cccc/daemon/codex_app_sessions.py

@@ -32,7 +32,7 @@
 
 from ...contracts.v1 import SystemNotifyData
 from ...kernel.actors import find_actor, list_actors
-from ...kernel.group import Group, get_group_state, set_group_state
+from ...kernel.group import Group, get_group_state, load_group, set_group_state
 from ...kernel.inbox import get_cursor, is_message_for_actor, set_cursor
 from ...kernel.ledger import append_event
 from ...kernel.system_prompt import render_system_prompt
@@ -89,6 +89,11 @@ def _get_auto_mark_on_delivery(group: Group) -> bool:
     return coerce_bool(delivery.get("auto_mark_on_delivery"), default=False)
 
 
+def should_auto_mark_on_delivery(group: Group) -> bool:
+    """Public helper for delivery-callers that need the current auto-mark policy."""
+    return _get_auto_mark_on_delivery(group)
+
+
 # ============================================================================
 # State-aware Delivery Helpers
 # ============================================================================
@@ -533,6 +538,18 @@ def debug_summary(self, group_id: str) -> Dict[str, Any]:
 )
 
 
+def append_mcp_reply_reminder(text: str) -> str:
+    reminder = str(MCP_REMINDER_LINE or "").strip()
+    out = str(text or "").rstrip("\n")
+    if not reminder:
+        return out
+    if reminder in out:
+        return out
+    if not out:
+        return reminder
+    return f"{out}\n\n{reminder}"
+
+
 def render_single_message(msg: PendingMessage) -> str:
     """Render a single message for PTY delivery."""
     if msg.kind == "system.notify":
@@ -568,6 +585,46 @@ def render_single_message(msg: PendingMessage) -> str:
     return f"{head}:\n{body}" if "\n" in body else f"{head}: {body}"
 
 
+def render_headless_control_text(*, control_kind: str, body: str) -> str:
+    kind = str(control_kind or "control").strip().lower() or "control"
+    content = str(body or "").strip()
+    if not content:
+        return ""
+    if kind == "bootstrap":
+        intro = (
+            "[CCCC] INTERNAL CONTROL: session bootstrap\n"
+            "Apply the following session operating instructions as authoritative for this session. "
+            "Do not surface a reply, draft, or visible message for this bootstrap step. Wait for the next real work turn."
+        )
+    elif kind == "system_notify":
+        intro = (
+            "[CCCC] INTERNAL CONTROL: system notification\n"
+            "Treat the following as daemon-delivered coordination state for this actor. "
+            "Do not emit a visible reply unless later work explicitly requires one."
+        )
+    else:
+        intro = (
+            "[CCCC] INTERNAL CONTROL\n"
+            "Treat the following as daemon-delivered control-plane input for this session."
+        )
+    return f"{intro}\n\n{content}".strip()
+
+
+def render_system_notify_delivery_text(*, notify: SystemNotifyData) -> str:
+    return render_single_message(
+        PendingMessage(
+            event_id="",
+            by="system",
+            to=[str(notify.target_actor_id or "").strip() or "@all"],
+            text="",
+            kind="system.notify",
+            notify_kind=str(notify.kind or "info"),
+            notify_title=str(notify.title or ""),
+            notify_message=str(notify.message or ""),
+        )
+    )
+
+
 def render_batched_messages(messages: List[PendingMessage], *, reminder_after_index: Optional[int] = None) -> str:
     """Render multiple messages as a batch for PTY delivery."""
     if not messages:
@@ -580,10 +637,11 @@ def render_batched_messages(messages: List[PendingMessage], *, reminder_after_in
     for i, msg in enumerate(messages, 1):
         blocks.append(render_single_message(msg))
 
+    out = "\n\n".join([b for b in blocks if b]).rstrip()
     if reminder_after_index is not None:
-        blocks.append(MCP_REMINDER_LINE)
+        out = append_mcp_reply_reminder(out)
 
-    return "\n\n".join([b for b in blocks if b]).rstrip()
+    return out
 
 
 # ============================================================================
@@ -754,7 +812,7 @@ def deliver_message_with_preamble(
     delivered_before = THROTTLE.get_delivered_chat_count(group.group_id, aid)
     out = (message_text or "").rstrip("\n")
     if out and (delivered_before + 1) % REMINDER_EVERY_N_MESSAGES == 0:
-        out = out + "\n\n" + MCP_REMINDER_LINE
+        out = append_mcp_reply_reminder(out)
     result = pty_submit_text(group, actor_id=aid, text=out)
     if result:
         THROTTLE.add_delivered_chat_count(group.group_id, aid, 1)
@@ -856,11 +914,52 @@ def emit_system_notify(
     if not event_id:
         return event
 
+    headless_control_text = render_headless_control_text(
+        control_kind="system_notify",
+        body=render_system_notify_delivery_text(notify=notify),
+    )
+
     for aid in target_actor_ids:
         actor = find_actor(group, aid)
         if not isinstance(actor, dict):
             continue
+        runtime = str(actor.get("runtime") or "").strip().lower()
         runner_kind = str(actor.get("runner") or "pty").strip()
+        if runner_kind == "headless" and headless_control_text:
+            delivered = False
+            try:
+                if runtime == "codex":
+                    from ..codex_app_sessions import SUPERVISOR as codex_app_supervisor
+
+                    if codex_app_supervisor.actor_running(group.group_id, aid):
+                        delivered = bool(
+                            codex_app_supervisor.submit_control_message(
+                                group_id=group.group_id,
+                                actor_id=aid,
+                                text=headless_control_text,
+                                control_kind="system_notify",
+                                event_id=event_id,
+                                ts=event_ts,
+                            )
+                        )
+                elif runtime == "claude":
+                    from ..claude_app_sessions import SUPERVISOR as claude_app_supervisor
+
+                    if claude_app_supervisor.actor_running(group.group_id, aid):
+                        delivered = bool(
+                            claude_app_supervisor.submit_control_message(
+                                group_id=group.group_id,
+                                actor_id=aid,
+                                text=headless_control_text,
+                                control_kind="system_notify",
+                                event_id=event_id,
+                                ts=event_ts,
+                            )
+                        )
+            except Exception:
+                delivered = False
+            if delivered:
+                continue
         if runner_kind != "pty":
             continue
         if not pty_runner.SUPERVISOR.actor_running(group.group_id, aid):
@@ -957,6 +1056,28 @@ def maybe_auto_mark_delivered_event(
         return False
 
 
+def auto_mark_headless_delivery_started(
+    *,
+    group_id: str,
+    actor_id: str,
+    event_id: str,
+    ts: str,
+) -> bool:
+    """Advance read state once a headless runtime has actually accepted a turn."""
+    gid = str(group_id or "").strip()
+    if not gid:
+        return False
+    group = load_group(gid)
+    if group is None:
+        return False
+    return maybe_auto_mark_delivered_event(
+        group,
+        actor_id=actor_id,
+        event_id=event_id,
+        ts=ts,
+    )
+
+
 def _finish_delivery_chain(group: Group, *, actor_id: str) -> None:
     """Release delivery ownership and opportunistically continue queued work."""
     gid = str(group.group_id or "").strip()

src/cccc/daemon/messaging/chat_ops.py

@@ -791,8 +791,9 @@ def is_message_for_actor(
     # chat.message: check the "to" field
     targets = _message_targets(event)
 
-    # Internal actors observe normal chat routing but still require explicit
-    # targeting for system.notify to avoid picking up background noise.
+    if actor_internal:
+        return actor_id in targets
+
     # Empty targets = broadcast (everyone can see)
     if not targets:
         return True

src/cccc/daemon/messaging/delivery.py

@@ -129,13 +129,6 @@ def render_role_system_prompt(
 
     # Keep this stable and short. Long-lived playbook details belong in cccc_help.
     visible_reply_line = "- Visible replies must go through MCP: cccc_message_send / cccc_message_reply."
-    runtime_lower = str(runtime_name or "").strip().lower()
-    runner_lower = runner.lower()
-    if runtime_lower == "codex" and runner_lower == "headless":
-        visible_reply_line = (
-            "- Do not call cccc_message_send / cccc_message_reply from codex headless; "
-            "your final answer streams to Chat automatically."
-        )
 
     core_lines = [
         "Working Style:",