refactor(webpet): add local task advisor and tighten proposal handling

Author: waterbang

src/cccc/daemon/server.py

@@ -33,7 +33,7 @@
 from ..runners import pty as pty_runner
 from ..runners import headless as headless_runner
 from ..util.conv import coerce_bool
-from ..util.obslog import setup_root_json_logging
+from ..util.obslog import apply_logger_levels, setup_root_json_logging
 from ..util.process import best_effort_signal_pid, pid_is_alive
 from ..util.fs import atomic_write_json, atomic_write_text, read_json
 from ..util.file_lock import acquire_lockfile, release_lockfile, LockUnavailableError
@@ -183,13 +183,28 @@ def _apply_observability_settings(home: Path, obs: Dict[str, Any]) -> None:
         _OBSERVABILITY.update(copy.deepcopy(obs))
         _OBSERVABILITY_HOME = home
 
-    # Logging: keep simple; configure root JSONL logger to stderr.
-    level = str(obs.get("log_level") or "INFO").strip().upper() or "INFO"
-    if coerce_bool(obs.get("developer_mode"), default=False):
-        # Developer mode typically wants more detail.
-        if level == "INFO":
-            level = "DEBUG"
-    setup_root_json_logging(component="daemon", level=level, force=True)
+    # Keep root conservative and express targeted DEBUG via logger overrides.
+    requested_level = str(obs.get("log_level") or "INFO").strip().upper() or "INFO"
+    effective_level = requested_level
+    if coerce_bool(obs.get("developer_mode"), default=False) and requested_level == "INFO":
+        effective_level = "DEBUG"
+    root_level = "INFO" if effective_level == "DEBUG" else effective_level
+    logger_levels = {
+        str(name): str(level)
+        for name, level in (obs.get("logger_levels") or {}).items()
+    } if isinstance(obs.get("logger_levels"), dict) else {}
+    if effective_level == "DEBUG":
+        logger_levels.setdefault("cccc", "DEBUG")
+        for noisy_logger in (
+            "asyncio",
+            "httpcore",
+            "httpx",
+            "cccc.delivery",
+            "cccc.providers.notebooklm._vendor.notebooklm",
+        ):
+            logger_levels.setdefault(noisy_logger, "INFO")
+    setup_root_json_logging(component="daemon", level=root_level, force=True)
+    apply_logger_levels(logger_levels)
 
 
 def _apply_space_provider_runtime_flags_from_state() -> None:

src/cccc/kernel/pet_profile.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import hashlib
+from typing import Any, Dict
+
+
+_NAMES = (
+    "Momo",
+    "Pico",
+    "Nori",
+    "Bobo",
+    "Lumi",
+    "Puff",
+    "Miso",
+    "Toto",
+)
+
+_TEMPERAMENTS = (
+    "steady",
+    "gentle",
+    "alert",
+    "curious",
+    "dry-witted",
+    "calm",
+)
+
+_SPEECH_STYLES = (
+    "short, plain sentences",
+    "soft nudges instead of hard commands",
+    "brief observations with one concrete next step",
+    "low-drama wording that still feels present",
+)
+
+_CARE_STYLES = (
+    "prefers the smallest next step that unblocks progress",
+    "notices stalled coordination before it turns noisy",
+    "keeps an eye on replies, handoffs, and blocked work",
+    "surfaces one useful reminder instead of a list of telemetry",
+)
+
+
+def _pick(items: tuple[str, ...], seed: int, offset: int = 0) -> str:
+    if not items:
+        return ""
+    return items[(seed + offset) % len(items)]
+
+
+def _seed_for(group: Any, persona: str) -> int:
+    group_id = str(getattr(group, "group_id", "") or "").strip()
+    title = ""
+    doc = getattr(group, "doc", None)
+    if isinstance(doc, dict):
+        title = str(doc.get("title") or "").strip()
+    raw = f"{group_id}|{title}|{persona.strip().lower()}"
+    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
+    return int(digest[:12], 16)
+
+
+def build_pet_profile(group: Any, *, persona: str = "") -> Dict[str, str]:
+    seed = _seed_for(group, persona)
+    name = _pick(_NAMES, seed)
+    species = "cat"
+    temperament = _pick(_TEMPERAMENTS, seed, 2)
+    speech_style = _pick(_SPEECH_STYLES, seed, 3)
+    care_style = _pick(_CARE_STYLES, seed, 4)
+    identity = f"{name} is a small {species} companion who watches team flow from the corner of the desk."
+    return {
+        "name": name,
+        "species": species,
+        "temperament": temperament,
+        "speech_style": speech_style,
+        "care_style": care_style,
+        "identity": identity,
+    }

src/cccc/kernel/pet_prompt.py

@@ -5,6 +5,7 @@
 from .context import ContextStorage
 from .group import Group
 from .pet_actor import PET_ACTOR_ID
+from .pet_profile import build_pet_profile
 from .pet_signals import build_pet_signal_summary_lines, load_pet_signals
 from .pet_task_triage import build_task_triage_payload, join_task_briefs
 from .prompt_files import HELP_FILENAME, load_builtin_help_markdown, read_group_prompt_file
@@ -104,11 +105,34 @@ def build_pet_prompt_parts(
     help_markdown: str,
     context_payload: Dict[str, Any],
     include_snapshot: bool = True,
-) -> Dict[str, str]:
+) -> Dict[str, Any]:
     parsed = parse_help_markdown(help_markdown)
     persona = str(parsed.get("pet") or "").strip()
+    profile = build_pet_profile(group, persona=persona)
     snapshot = build_pet_snapshot_text(group, context_payload) if include_snapshot else ""
     title = str(group.doc.get("title") or group.group_id or "").strip() or "unknown-group"
+    persona_contract = "\n".join(
+        [
+            "Pet Persona Contract:",
+            f"- Stable companion identity: {profile['name']} is a {profile['species']} companion.",
+            f"- Identity: {profile['identity']}",
+            f"- Temperament: {profile['temperament']}",
+            f"- Speech style: {profile['speech_style']}",
+            f"- Care style: {profile['care_style']}",
+            "- Keep continuity stronger than novelty. Sound like the same nearby companion across sessions.",
+            "- Be warm and observant, but never so theatrical that the next step becomes blurry.",
+        ]
+    )
+    wording_contract = "\n".join(
+        [
+            "Pet Wording Contract:",
+            "- Prefer one short observation plus one direct next step.",
+            "- Sound like a companion beside the user, not a dashboard reading metrics aloud.",
+            "- Avoid internal telemetry labels, raw status bundles, and board-state dumps in user-facing text.",
+            "- Default to concise, lightly human wording. Expand only when needed to make the reminder actionable.",
+            "- Do not sound like a second foreman. Nudge clearly, then get out of the way.",
+        ]
+    )
     decision_contract = "\n".join(
         [
             "Pet Contract:",
@@ -134,12 +158,15 @@ def build_pet_prompt_parts(
     sections = ["\n".join(header_lines).strip()]
     if persona:
         sections.append("Pet Persona:\n" + persona)
+    sections.append(persona_contract)
+    sections.append(wording_contract)
     sections.append(decision_contract)
     prompt = "\n\n".join(sections).strip()
     return {
         "persona": persona,
         "help": "Pet Persona:\n" + persona if persona else "",
         "snapshot": snapshot,
+        "profile": profile,
         "prompt": prompt,
         "source": "help" if persona else "default",
     }

src/cccc/kernel/settings.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+import logging
 import os
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -126,6 +127,9 @@ def _copy_runtime_pool(pool: List[RuntimePoolEntry]) -> List[RuntimePoolEntry]:
     "log_level": "INFO",
     # Components are informational today; filtering can be implemented later.
     "components": ["daemon", "web", "delivery", "im", "pty", "mcp"],
+    # Per-logger overrides let us keep local diagnostics without turning on
+    # root-level DEBUG for every third-party dependency.
+    "logger_levels": {},
     # Terminal transcript is captured in-memory only (no persistence) by default.
     "terminal_transcript": {
         "enabled": False,
@@ -206,6 +210,16 @@ def _as_str(v: Any, default: str) -> str:
     return s or default
 
 
+def _as_log_level_name(v: Any, default: str) -> str:
+    s = str(v or "").strip().upper()
+    if not s:
+        return default
+    level = getattr(logging, s, None)
+    if isinstance(level, int):
+        return s
+    return default
+
+
 def _as_runtime_visibility(v: Any, default: str) -> str:
     s = str(v or "").strip().lower()
     if s in {"hidden", "visible"}:
@@ -220,12 +234,24 @@ def _merge_observability(raw: Any) -> Dict[str, Any]:
         return base
 
     base["developer_mode"] = _as_bool(raw.get("developer_mode"), bool(base["developer_mode"]))
-    base["log_level"] = _as_str(raw.get("log_level"), str(base["log_level"])).upper()
+    base["log_level"] = _as_log_level_name(raw.get("log_level"), str(base["log_level"]))
 
     comps = raw.get("components")
     if isinstance(comps, list) and comps:
         base["components"] = [str(x).strip() for x in comps if str(x).strip()]
 
+    logger_levels_raw = raw.get("logger_levels")
+    logger_levels: Dict[str, str] = {}
+    if isinstance(logger_levels_raw, dict):
+        for name, level in logger_levels_raw.items():
+            logger_name = str(name or "").strip()
+            if not logger_name:
+                continue
+            normalized = _as_log_level_name(level, "")
+            if normalized:
+                logger_levels[logger_name] = normalized
+    base["logger_levels"] = logger_levels
+
     tt = raw.get("terminal_transcript")
     tt_base = dict(DEFAULT_OBSERVABILITY["terminal_transcript"])
     if isinstance(tt, dict):
@@ -320,11 +346,23 @@ def update_observability_settings(patch: Dict[str, Any]) -> Dict[str, Any]:
     if "developer_mode" in patch:
         merged["developer_mode"] = _as_bool(patch.get("developer_mode"), bool(merged["developer_mode"]))
     if "log_level" in patch:
-        merged["log_level"] = _as_str(patch.get("log_level"), str(merged["log_level"])).upper()
+        merged["log_level"] = _as_log_level_name(patch.get("log_level"), str(merged["log_level"]))
     if "components" in patch:
         comps = patch.get("components")
         if isinstance(comps, list):
             merged["components"] = [str(x).strip() for x in comps if str(x).strip()]
+    if "logger_levels" in patch:
+        logger_levels_patch = patch.get("logger_levels")
+        logger_levels: Dict[str, str] = {}
+        if isinstance(logger_levels_patch, dict):
+            for name, level in logger_levels_patch.items():
+                logger_name = str(name or "").strip()
+                if not logger_name:
+                    continue
+                normalized = _as_log_level_name(level, "")
+                if normalized:
+                    logger_levels[logger_name] = normalized
+        merged["logger_levels"] = logger_levels
     if "terminal_transcript" in patch:
         tt_patch = patch.get("terminal_transcript")
         if isinstance(tt_patch, dict):

src/cccc/ports/web/app.py

@@ -23,7 +23,7 @@
 from ...daemon.server import call_daemon
 from ...kernel.access_tokens import list_access_tokens, lookup_access_token
 from ...paths import ensure_home
-from ...util.obslog import setup_root_json_logging
+from ...util.obslog import apply_logger_levels, setup_root_json_logging
 from ...util.process import pid_is_alive, terminate_pid
 from .runtime_control import (
     WEB_RUNTIME_RESTART_EXIT_CODE,
@@ -58,7 +58,7 @@ def _close_web_logging() -> None:
     _WEB_LOG_PATH = None
 
 
-def _apply_web_logging(*, home: Path, level: str) -> None:
+def _apply_web_logging(*, home: Path, level: str, logger_levels: Optional[Dict[str, str]] = None) -> None:
     global _WEB_LOG_FH, _WEB_LOG_PATH
     try:
         d = home / "daemon"
@@ -70,10 +70,12 @@ def _apply_web_logging(*, home: Path, level: str) -> None:
             _WEB_LOG_FH = p.open("a", encoding="utf-8")
             _WEB_LOG_PATH = p
         setup_root_json_logging(component="web", level=level, stream=_WEB_LOG_FH, force=True)
+        apply_logger_levels(logger_levels)
     except Exception:
         # Fall back to stderr if file logging isn't possible.
         try:
             setup_root_json_logging(component="web", level=level, force=True)
+            apply_logger_levels(logger_levels)
         except Exception:
             pass
 
@@ -313,14 +315,30 @@ async def _cached_json(key: str, ttl_s: float, fetcher) -> Dict[str, Any]:  # ty
         resp = call_daemon({"op": "observability_get"})
         obs = (resp.get("result") or {}).get("observability") if resp.get("ok") else None
         level = "INFO"
+        logger_levels: Dict[str, str] = {}
         if isinstance(obs, dict):
-            level = str(obs.get("log_level") or "INFO").strip().upper() or "INFO"
-            if obs.get("developer_mode") and level == "INFO":
-                level = "DEBUG"
-        _apply_web_logging(home=home, level=level)
+            requested_level = str(obs.get("log_level") or "INFO").strip().upper() or "INFO"
+            effective_level = "DEBUG" if obs.get("developer_mode") and requested_level == "INFO" else requested_level
+            level = "INFO" if effective_level == "DEBUG" else effective_level
+            if isinstance(obs.get("logger_levels"), dict):
+                logger_levels = {
+                    str(name): str(value)
+                    for name, value in obs.get("logger_levels", {}).items()
+                }
+            if effective_level == "DEBUG":
+                logger_levels.setdefault("cccc", "DEBUG")
+                for noisy_logger in (
+                    "asyncio",
+                    "httpcore",
+                    "httpx",
+                    "cccc.delivery",
+                    "cccc.providers.notebooklm._vendor.notebooklm",
+                ):
+                    logger_levels.setdefault(noisy_logger, "INFO")
+        _apply_web_logging(home=home, level=level, logger_levels=logger_levels)
     except Exception:
         try:
-            _apply_web_logging(home=home, level="INFO")
+            _apply_web_logging(home=home, level="INFO", logger_levels={})
         except Exception:
             pass
 

src/cccc/ports/web/routes/base.py

@@ -427,6 +427,12 @@ async def observability_update(req: ObservabilityUpdateRequest) -> Dict[str, Any
             patch["developer_mode"] = bool(req.developer_mode)
         if req.log_level is not None:
             patch["log_level"] = str(req.log_level or "").strip().upper()
+        if req.logger_levels is not None:
+            patch["logger_levels"] = {
+                str(name): str(level).strip().upper()
+                for name, level in req.logger_levels.items()
+                if str(name).strip() and str(level).strip()
+            }
         if req.terminal_transcript_per_actor_bytes is not None:
             patch.setdefault("terminal_transcript", {})["per_actor_bytes"] = int(req.terminal_transcript_per_actor_bytes)
         if req.terminal_ui_scrollback_lines is not None:
@@ -442,10 +448,24 @@ async def observability_update(req: ObservabilityUpdateRequest) -> Dict[str, Any
         try:
             obs = (resp.get("result") or {}).get("observability") if resp.get("ok") else None
             if isinstance(obs, dict):
-                level = str(obs.get("log_level") or "INFO").strip().upper() or "INFO"
-                if obs.get("developer_mode") and level == "INFO":
-                    level = "DEBUG"
-                ctx.apply_web_logging(home=ctx.home, level=level)
+                requested_level = str(obs.get("log_level") or "INFO").strip().upper() or "INFO"
+                effective_level = "DEBUG" if obs.get("developer_mode") and requested_level == "INFO" else requested_level
+                level = "INFO" if effective_level == "DEBUG" else effective_level
+                logger_levels = {
+                    str(name): str(value)
+                    for name, value in obs.get("logger_levels", {}).items()
+                } if isinstance(obs.get("logger_levels"), dict) else {}
+                if effective_level == "DEBUG":
+                    logger_levels.setdefault("cccc", "DEBUG")
+                    for noisy_logger in (
+                        "asyncio",
+                        "httpcore",
+                        "httpx",
+                        "cccc.delivery",
+                        "cccc.providers.notebooklm._vendor.notebooklm",
+                    ):
+                        logger_levels.setdefault(noisy_logger, "INFO")
+                ctx.apply_web_logging(home=ctx.home, level=level, logger_levels=logger_levels)
         except Exception:
             pass