fix: tighten truth boundaries across web and pet flows

- require canonical readback for task and context writeback paths
- distinguish accepted versus completed group space operations
- make pet review and profile refresh complete on real assistive outputs

Author: ChesterRa

src/cccc/daemon/context/context_ops.py

@@ -43,6 +43,7 @@
 from ...kernel.actors import get_effective_role, list_actors
 from ...kernel.group import load_group
 from ...kernel.query_projections import get_actor_list_projection
+from ...kernel.pet_actor import PET_ACTOR_ID
 from ...kernel.ledger import append_event
 from ...kernel.prompt_files import (
     HELP_FILENAME,
@@ -77,6 +78,43 @@ def _error(code: str, message: str, *, details: Optional[Dict[str, Any]] = None)
     return DaemonResponse(ok=False, error=DaemonError(code=code, message=message, details=(details or {})))
 
 
+_PET_FORBIDDEN_CONTEXT_OPS = {
+    "coordination.brief.update",
+    "coordination.note.add",
+    "task.create",
+    "task.update",
+    "task.move",
+    "task.restore",
+    "task.delete",
+    "meta.merge",
+    "role_notes.set",
+    "agent_state.clear",
+}
+
+
+def _check_pet_context_permission(by: str, op_name: str, *, target_actor_id: Optional[str] = None) -> Optional[str]:
+    if str(by or "").strip() != PET_ACTOR_ID:
+        return None
+    if op_name in _PET_FORBIDDEN_CONTEXT_OPS:
+        return f"Permission denied: pet cannot run {op_name}"
+    if op_name == "agent_state.update":
+        target = str(target_actor_id or "").strip()
+        if target and target != PET_ACTOR_ID:
+            return f"Permission denied: pet can only update its own agent_state ({PET_ACTOR_ID})"
+    return None
+
+
+def _validate_pet_agent_state_update(raw: Dict[str, Any], *, actor_id: str) -> Optional[str]:
+    target = str(actor_id or "").strip()
+    if target != PET_ACTOR_ID:
+        return f"Permission denied: pet can only update its own agent_state ({PET_ACTOR_ID})"
+    allowed = {"op", "actor_id", "agent_id", "user_model", "user_profile"}
+    extra = sorted(key for key in raw.keys() if key not in allowed)
+    if extra:
+        return "Permission denied: pet agent_state.update only allows user_model; disallowed keys: " + ", ".join(extra)
+    return None
+
+
 def _status_value(value: Any) -> str:
     if isinstance(value, Enum):
         return str(value.value or "")
@@ -545,6 +583,10 @@ def _check_permission(
     if group is None:
         return None
 
+    pet_err = _check_pet_context_permission(by, op_name, target_actor_id=target_actor_id)
+    if pet_err:
+        return pet_err
+
     role = "user" if by == "user" else get_effective_role(group, by)
     if role in {"user", "foreman"}:
         if op_name in {"agent_state.update", "agent_state.clear"} and target_actor_id and by not in {"system", target_actor_id}:
@@ -1503,6 +1545,10 @@ def _mark_change(index: int, op_name: str, detail: str, *, task_id: str = "") ->
                 perm_err = _check_permission(by, op_name, group_id, target_actor_id=actor_id)
                 if perm_err:
                     raise ValueError(perm_err)
+                if by == PET_ACTOR_ID:
+                    pet_update_err = _validate_pet_agent_state_update(raw, actor_id=actor_id)
+                    if pet_update_err:
+                        raise ValueError(pet_update_err)
                 agent = _get_or_create_agent(agents_state, actor_id)
                 updated = False
                 hot = agent.hot if isinstance(agent.hot, AgentStateHot) else AgentStateHot()
@@ -1583,6 +1629,10 @@ def _mark_change(index: int, op_name: str, detail: str, *, task_id: str = "") ->
                 perm_err = _check_permission(by, op_name, group_id, target_actor_id=actor_id)
                 if perm_err:
                     raise ValueError(perm_err)
+                if by == PET_ACTOR_ID:
+                    pet_update_err = _validate_pet_agent_state_update(raw, actor_id=actor_id)
+                    if pet_update_err:
+                        raise ValueError(pet_update_err)
                 agent = _get_or_create_agent(agents_state, actor_id)
                 agent.hot = AgentStateHot()
                 agent.warm = AgentStateWarm()

src/cccc/daemon/messaging/chat_ops.py

@@ -20,6 +20,7 @@
     targets_any_agent,
 )
 from ...kernel.scope import detect_scope
+from ...kernel.pet_actor import PET_ACTOR_ID, get_pet_actor
 from ...util.time import utc_now_iso
 from .delivery import (
     flush_pending_messages,
@@ -37,6 +38,13 @@ def _error(code: str, message: str, *, details: Optional[Dict[str, Any]] = None)
     return DaemonResponse(ok=False, error=DaemonError(code=code, message=message, details=(details or {})))
 
 
+def _is_internal_pet_sender(group: Any, by: str) -> bool:
+    actor_id = str(by or "").strip()
+    if actor_id != PET_ACTOR_ID:
+        return False
+    return isinstance(get_pet_actor(group), dict)
+
+
 def _wake_group_on_human_message(
     group: Any,
     *,
@@ -328,6 +336,11 @@ def handle_send(
     group = load_group(group_id)
     if group is None:
         return _error("group_not_found", f"group not found: {group_id}")
+    if _is_internal_pet_sender(group, by):
+        return _error(
+            "pet_visible_chat_forbidden",
+            "Pet cannot send or reply visible chat directly; use pet decisions instead.",
+        )
 
     group = _wake_group_on_human_message(
         group,
@@ -546,6 +559,11 @@ def handle_reply(
     group = load_group(group_id)
     if group is None:
         return _error("group_not_found", f"group not found: {group_id}")
+    if _is_internal_pet_sender(group, by):
+        return _error(
+            "pet_visible_chat_forbidden",
+            "Pet cannot send or reply visible chat directly; use pet decisions instead.",
+        )
 
     group = _wake_group_on_human_message(
         group,