devplacepy/devplacepy/services/bot/service.py

# retoor <retoor@molodetz.nl>

import asyncio
import importlib
import logging
from collections import deque
from datetime import datetime, timedelta, timezone

from devplacepy.services.base import BaseService, ConfigField
from devplacepy.services.bot import config
from devplacepy.services.bot.monitor import monitor

logger = logging.getLogger(__name__)


class BotsService(BaseService):
    default_enabled = False
    min_interval = 5
    title = "Bots"
    description = (
        "Runs a fleet of Playwright-driven AI personas that browse a DevPlace "
        "instance and post, comment, vote, and react like real users. Fleet size, "
        "target site, and LLM settings are configurable, with live cost and usage "
        "metrics per bot."
    )
    config_fields = [
        ConfigField(
            "bot_fleet_size",
            "Fleet size (bots)",
            type="int",
            default=1,
            minimum=0,
            maximum=20,
            help="Number of concurrent bot accounts to run.",
            group="Fleet",
        ),
        ConfigField(
            "bot_headless",
            "Headless browser",
            type="bool",
            default=True,
            help="Run Chromium headless. Disable only for local debugging on a desktop.",
            group="Fleet",
        ),
        ConfigField(
            "bot_max_actions",
            "Max actions per bot",
            type="int",
            default=0,
            minimum=0,
            help="Stop a bot after this many actions (0 = run indefinitely).",
            group="Fleet",
        ),
        ConfigField(
            "bot_base_url",
            "Target site URL",
            type="url",
            default=config.BASE_URL_DEFAULT,
            help="DevPlace instance the bots browse and post to.",
            group="Target",
        ),
        ConfigField(
            "bot_news_api",
            "News API URL",
            type="url",
            default=config.NEWS_API_DEFAULT,
            help="Source of articles the bots react to.",
            group="Target",
        ),
        ConfigField(
            "bot_api_url",
            "LLM API URL",
            type="url",
            default=config.API_URL_DEFAULT,
            help="OpenAI-compatible chat-completions endpoint.",
            group="LLM",
        ),
        ConfigField(
            "bot_model",
            "LLM model",
            type="str",
            default=config.MODEL_DEFAULT,
            help="Model name sent to the LLM API.",
            group="LLM",
        ),
        ConfigField(
            "bot_api_key",
            "LLM API key",
            type="password",
            default="",
            secret=True,
            help="Defaults to the gateway's internal key (the bots use the local AI gateway).",
            group="LLM",
        ),
        ConfigField(
            "bot_input_cost_per_1m",
            "Input cost per 1M tokens ($)",
            type="float",
            default=config.INPUT_COST_PER_1M_DEFAULT,
            minimum=0,
            help="Used to compute live spend per bot and across the fleet.",
            group="LLM",
        ),
        ConfigField(
            "bot_output_cost_per_1m",
            "Output cost per 1M tokens ($)",
            type="float",
            default=config.OUTPUT_COST_PER_1M_DEFAULT,
            minimum=0,
            help="Used to compute live spend per bot and across the fleet.",
            group="LLM",
        ),
        ConfigField(
            "bot_max_per_article",
            "Bots per article",
            type="int",
            default=config.MAX_BOTS_PER_ARTICLE,
            minimum=1,
            maximum=5,
            help="How many bots may post about the same news article, each from a different angle.",
            group="Behavior",
        ),
        ConfigField(
            "bot_article_ttl_days",
            "Article cooldown (days)",
            type="int",
            default=config.ARTICLE_TTL_DAYS,
            minimum=1,
            maximum=90,
            help="How long an article stays covered before it can be posted about again.",
            group="Behavior",
        ),
        ConfigField(
            "bot_gist_min_lines",
            "Minimum gist lines",
            type="int",
            default=config.GIST_MIN_LINES,
            minimum=1,
            maximum=50,
            help="Reject generated code snippets shorter than this many non-empty lines.",
            group="Behavior",
        ),
        ConfigField(
            "bot_action_pause_min_seconds",
            "Min pause between actions (s)",
            type="int",
            default=config.ACTION_PAUSE_MIN_SECONDS,
            minimum=0,
            maximum=3600,
            help="Lower bound of the idle pause a bot takes after each action.",
            group="Pacing",
        ),
        ConfigField(
            "bot_action_pause_max_seconds",
            "Max pause between actions (s)",
            type="int",
            default=config.ACTION_PAUSE_MAX_SECONDS,
            minimum=1,
            maximum=7200,
            help="Upper bound of the idle pause a bot takes after each action.",
            group="Pacing",
        ),
        ConfigField(
            "bot_break_scale",
            "Break length multiplier",
            type="float",
            default=config.BREAK_SCALE_DEFAULT,
            minimum=0.1,
            maximum=10.0,
            help="Scales the between-session break. Below 1 makes the fleet more active (and costlier); above 1 calmer.",
            group="Pacing",
        ),
        ConfigField(
            "bot_startup_jitter_seconds",
            "Startup jitter (s)",
            type="int",
            default=config.STARTUP_JITTER_SECONDS_DEFAULT,
            minimum=0,
            maximum=86400,
            help="Each bot waits a random delay up to this many seconds before its first session, spreading fleet activity instead of bursting together when the service starts.",
            group="Pacing",
        ),
        ConfigField(
            "bot_ai_decisions",
            "AI-driven decisions",
            type="bool",
            default=config.AI_DECISIONS_DEFAULT,
            help="Let each bot's persona decide every action via the LLM instead of fixed probabilities. Adds one decision call per page; falls back to the procedural cascade when off.",
            group="Decisions",
        ),
        ConfigField(
            "bot_decision_temperature",
            "Decision temperature",
            type="float",
            default=config.DECISION_TEMPERATURE_DEFAULT,
            minimum=0.0,
            maximum=2.0,
            help="Sampling temperature for the per-page decision call. Low values keep structured output stable; variety comes from the identity, not noise.",
            group="Decisions",
        ),
    ]

    def __init__(self):
        super().__init__(name="bots", interval_seconds=15)
        self._fleet = {}
        self._state_dir = config.STATE_DIR
        self._cost_samples = deque()
        self._cost_anchor = None

    def _resolve_api_key(self, cfg: dict) -> str:
        from devplacepy.database import internal_gateway_key

        return cfg["bot_api_key"] or internal_gateway_key()

    @staticmethod
    def _ensure_importable() -> None:
        importlib.import_module("devplacepy.services.bot.bot")

    async def run_once(self) -> None:
        cfg = self.get_config()
        desired = cfg["bot_fleet_size"]
        api_key = self._resolve_api_key(cfg)
        if not api_key:
            if self._fleet:
                await self._stop_fleet()
            self.log(
                "No LLM API key available (bot_api_key / gateway internal key); fleet idle"
            )
            return
        try:
            self._ensure_importable()
        except ImportError as e:
            if self._fleet:
                await self._stop_fleet()
            self.log(f"Bot dependencies unavailable ({e}); install the 'bots' extra")
            return

        for slot in list(self._fleet):
            task = self._fleet[slot]["task"]
            if task.done():
                self._report_exit(slot, task)
                del self._fleet[slot]

        while len(self._fleet) > desired:
            await self._stop_slot(max(self._fleet))

        for slot in range(desired):
            if slot not in self._fleet:
                self._launch_slot(slot, cfg, api_key)

        self.log(f"Fleet: {len(self._fleet)}/{desired} bots active")

    def _report_exit(self, slot: int, task) -> None:
        if task.cancelled():
            return
        exc = task.exception()
        if exc:
            self.log(f"bot{slot} exited with error: {exc}; relaunching")
        else:
            self.log(f"bot{slot} finished; relaunching")

    def _launch_slot(self, slot: int, cfg: dict, api_key: str) -> None:
        from devplacepy.services.bot.bot import DevPlaceBot
        from devplacepy.services.bot.runtime import BotRuntimeConfig

        rc = BotRuntimeConfig(
            state_path=self._state_dir / f"state_slot{slot}.json",
            base_url=cfg["bot_base_url"],
            api_url=cfg["bot_api_url"],
            news_api=cfg["bot_news_api"],
            model=cfg["bot_model"],
            api_key=api_key,
            input_cost_per_1m=cfg["bot_input_cost_per_1m"],
            output_cost_per_1m=cfg["bot_output_cost_per_1m"],
            headless=cfg["bot_headless"],
            registry_path=config.ARTICLE_REGISTRY_PATH,
            max_per_article=cfg["bot_max_per_article"],
            article_ttl_days=cfg["bot_article_ttl_days"],
            gist_min_lines=cfg["bot_gist_min_lines"],
            action_pause_min=cfg["bot_action_pause_min_seconds"],
            action_pause_max=cfg["bot_action_pause_max_seconds"],
            break_scale=cfg["bot_break_scale"],
            startup_jitter_seconds=cfg["bot_startup_jitter_seconds"],
            ai_decisions=cfg["bot_ai_decisions"],
            decision_temperature=cfg["bot_decision_temperature"],
        )
        bot = DevPlaceBot(rc, on_event=self.log, slot=slot)
        task = asyncio.create_task(self._run_slot(slot, bot, cfg["bot_max_actions"]))
        self._fleet[slot] = {"bot": bot, "task": task}
        self.log(f"Launched bot{slot}")

    async def _run_slot(self, slot: int, bot, max_actions: int) -> None:
        try:
            await bot.run_forever(action_limit=max_actions)
        except asyncio.CancelledError:
            raise
        except Exception as e:
            self.log(f"bot{slot} crashed: {e}")

    async def on_disable(self) -> None:
        await self._stop_fleet()

    async def _stop_fleet(self) -> None:
        slots = list(self._fleet)
        if slots:
            await asyncio.gather(
                *(self._stop_slot(slot) for slot in slots), return_exceptions=True
            )

    async def _stop_slot(self, slot: int) -> None:
        entry = self._fleet.pop(slot, None)
        if not entry:
            return
        task = entry["task"]
        task.cancel()
        try:
            await asyncio.wait_for(task, timeout=30)
        except (asyncio.CancelledError, asyncio.TimeoutError):
            pass
        except Exception as e:
            self.log(f"bot{slot} stop error: {e}")
        await self._close_bot(slot, entry.get("bot"))
        monitor.drop(slot)
        self.log(f"Stopped bot{slot}")

    async def _close_bot(self, slot: int, bot) -> None:
        browser = getattr(bot, "b", None)
        if browser is None:
            return
        try:
            await asyncio.wait_for(browser.close(), timeout=15)
        except (asyncio.CancelledError, asyncio.TimeoutError):
            pass
        except Exception as e:
            self.log(f"bot{slot} browser close error: {e}")

    def collect_metrics(self) -> dict:
        rows = []
        total_cost = 0.0
        total_calls = total_in = total_out = 0
        posts = comments = votes = running = 0
        for slot in sorted(self._fleet):
            entry = self._fleet[slot]
            bot = entry["bot"]
            alive = not entry["task"].done()
            running += 1 if alive else 0
            llm = getattr(bot, "llm", None)
            st = getattr(bot, "state", None)
            cost = llm.total_cost if llm else 0.0
            calls = llm.total_calls if llm else 0
            total_cost += cost
            total_calls += calls
            total_in += llm.total_in_tokens if llm else 0
            total_out += llm.total_out_tokens if llm else 0
            if st:
                posts += st.created_posts
                comments += st.comments_posted
                votes += st.votes_cast
            username = (st.username if st else "") or ""
            user_cell = (
                {
                    "text": username,
                    "href": f"/profile/{username}",
                    "avatar": username,
                }
                if username
                else "-"
            )
            rows.append(
                [
                    f"bot{slot}",
                    user_cell,
                    (st.persona if st else "") or "-",
                    "running" if alive else "stopped",
                    st.created_posts if st else 0,
                    st.comments_posted if st else 0,
                    st.votes_cast if st else 0,
                    calls,
                    f"${cost:.4f}",
                ]
            )
        window, observed_cost, ready = self._sample_cost(
            datetime.now(timezone.utc), total_cost
        )
        cost_per_hour = observed_cost / window * 3600 if ready else 0.0
        projected_24h = observed_cost / window * 86400 if ready else 0.0
        rate_value = f"${cost_per_hour:.4f}/h" if ready else "warming up"
        projected_value = f"${projected_24h:.2f}" if ready else "warming up"
        stats = [
            {"label": "Bots running", "value": running},
            {"label": "Fleet cost", "value": f"${total_cost:.4f}"},
            {"label": "Observed window", "value": f"{window:.0f}s"},
            {"label": "Cost rate", "value": rate_value},
            {"label": "Projected 24h cost", "value": projected_value},
            {"label": "LLM calls", "value": total_calls},
            {"label": "Tokens in", "value": total_in},
            {"label": "Tokens out", "value": total_out},
            {"label": "Posts", "value": posts},
            {"label": "Comments", "value": comments},
            {"label": "Votes", "value": votes},
        ]
        table = {
            "columns": [
                "Bot",
                "User",
                "Persona",
                "Status",
                "Posts",
                "Comments",
                "Votes",
                "Calls",
                "Cost",
            ],
            "rows": rows,
        }
        return {"stats": stats, "table": table, "frames": monitor.snapshot()}

    def _sample_cost(
        self, now: datetime, total_cost: float
    ) -> tuple[float, float, bool]:
        if self._started_at is None:
            self._cost_samples.clear()
            self._cost_anchor = None
            return 0.0, 0.0, False
        if self._cost_anchor != self._started_at:
            self._cost_samples.clear()
            self._cost_anchor = self._started_at
        self._cost_samples.append((now, total_cost))
        cutoff = now - timedelta(seconds=config.COST_WINDOW_SECONDS)
        while len(self._cost_samples) > 2 and self._cost_samples[0][0] < cutoff:
            self._cost_samples.popleft()
        start_at, start_cost = self._cost_samples[0]
        window = (now - start_at).total_seconds()
        observed_cost = max(0.0, total_cost - start_cost)
        ready = window >= config.COST_WARMUP_SECONDS
        return window, observed_cost, ready