fix(#29): harden bot lifecycle, persona-correct scheduled jobs, redact tokens

Adversarial review fixes:
- main: one bad/duplicate persona token no longer aborts startup — each bot
  starts independently; tokens are deduped; partial-failure tears down already
  -started bots so none are left orphaned polling. Shutdown stops every bot
  independently (one failure can't strand the rest).
- scheduler/agent: a telegram:<persona> job is generated AS that persona via a
  new process(persona_name=…) override, keeping the 'system' execution mode
  (auto-approved writes, no memory/reflection) — fixes jobs being written in
  the default identity.
- admin: persona bot_token is redacted on read (like the global token) and not
  leaked via the raw markdown view.
- telegram: persona bots skip topic auto-bind (rung 0 ignores it) and the global
  browser-status mirror (no cross-bot progress spam).

Author: mattmezza

api/admin.py

@@ -481,6 +481,21 @@ class SkillUpsertIn(BaseModel):
     content: str
 
 
+def _persona_public(persona) -> dict:
+    """Persona as JSON for read-only APIs, with the bot token redacted (#29).
+
+    Mirrors how the global Telegram token is redacted in config reads — the token
+    is a secret and must not leave the server in cleartext (exports, list views).
+    """
+    from dataclasses import asdict, replace
+
+    from core.config_store import _redact
+    from core.personae import to_markdown
+
+    safe = replace(persona, bot_token=_redact(persona.bot_token))
+    return {**asdict(safe), "markdown": to_markdown(safe)}
+
+
 class PersonaUpsertIn(BaseModel):
     name: str
     agent_name: str = ""
@@ -2271,30 +2286,22 @@ async def _personae_partial() -> HTMLResponse:
 
     @app.get("/personae", dependencies=[Depends(auth)])
     async def list_personae() -> dict:
-        from dataclasses import asdict
-
-        from core.personae import to_markdown
-
         store = await _persona_store_from_config(config_store)
         personae = await store.list_personae()
         active = (await config_store.get("agent.active_persona") or "").strip()
         return {
             "count": len(personae),
             "active": active,
-            "personae": [{**asdict(p), "markdown": to_markdown(p)} for p in personae],
+            "personae": [_persona_public(p) for p in personae],
         }
 
     @app.get("/personae/{name}", dependencies=[Depends(auth)])
     async def get_persona(name: str) -> dict:
-        from dataclasses import asdict
-
-        from core.personae import to_markdown
-
         store = await _persona_store_from_config(config_store)
         persona = await store.get(name)
         if not persona:
             raise HTTPException(404, f"Persona not found: {name}")
-        return {**asdict(persona), "markdown": to_markdown(persona)}
+        return _persona_public(persona)
 
     @app.post("/personae", dependencies=[Depends(auth)])
     async def upsert_persona(body: PersonaUpsertIn) -> HTMLResponse:

channels/telegram.py

@@ -57,7 +57,10 @@ def __init__(
         self.app.add_handler(MessageHandler(filters.TEXT, self._on_text))
         self.app.add_handler(MessageHandler(filters.VOICE | filters.AUDIO, self._on_voice))
         self.app.add_handler(MessageHandler(filters.PHOTO | filters.Document.IMAGE, self._on_photo))
-        if config.topics_enabled:
+        # Topic→persona auto-bind only makes sense on the default bot: a persona
+        # bot resolves straight to its own persona (rung 0), so a per-topic binding
+        # would be ignored. Topic *folding* (history isolation) still applies below.
+        if config.topics_enabled and channel_name == "telegram":
             self.app.add_handler(
                 MessageHandler(
                     filters.StatusUpdate.FORUM_TOPIC_CREATED
@@ -378,6 +381,13 @@ async def _progress(self, chat_id: int | str):
         poll it and edit a single Telegram message in place (the chat equivalent
         of the REPL's self-updating spinner line). No-op when nothing is running.
         """
+        # ponytail: the explore status file is a single global singleton, so only
+        # the default bot mirrors it — otherwise a run triggered via one persona-bot
+        # would bubble into every other bot's chat (#29). Per-run scoping (a status
+        # path keyed by channel/profile) belongs in the browser tool — follow-up.
+        if self.channel_name != "telegram":
+            yield
+            return
         status = Path("/app/data" if Path("/app/data").exists() else "data")
         status = status / "browser" / "last" / "explore.status"
         cid, kw = self._route(chat_id)  # split a folded "<chat>:<thread>" topic id

core/agent.py

@@ -363,13 +363,19 @@ async def process(
         user_id: str,
         attachments: list[Attachment] | None = None,
         chat_id: str = "",
+        persona_name: str | None = None,
     ) -> AgentResponse:
         """Process an incoming message through the LLM with tool-use loop.
 
         ``chat_id`` distinguishes different chats for the same user (e.g.
         a private Telegram chat vs. a group chat).  Each unique
         (channel, user_id, chat_id) triple gets its own conversation history,
         preventing context leakage across chats.
+
+        ``persona_name`` forces the identity instead of resolving it from the
+        channel/binding ladder — used by the scheduler so a ``telegram:<persona>``
+        job is generated *as* that persona while keeping the ``system`` execution
+        mode (auto-approved writes, no memory/reflection) (#29).
         """
 
         # Handle /new (alias /clear) command — clear conversational context.
@@ -397,8 +403,12 @@ async def process(
         preamble = self._turn_preamble(decomposed_goal)
 
         # Resolve the active persona (its identity, skills + tool scope) — a
-        # per-chat binding wins over the globally selected persona (#14).
-        persona = await self._resolve_persona(channel, user_id, chat_id)
+        # per-chat binding wins over the globally selected persona (#14). An
+        # explicit override (scheduler) skips the ladder (#29).
+        if persona_name:
+            persona = await self._load_persona(persona_name)
+        else:
+            persona = await self._resolve_persona(channel, user_id, chat_id)
         tools = scoped_tools(persona)
         if self.secret_store is None:
             tools = [t for t in tools if t["name"] not in ("list_secrets", "request_secret")]

core/main.py

@@ -84,52 +84,87 @@ async def _start_agent(config_store: ConfigStore):
         )
         agent.voice = voice
 
-    # -- Telegram --
-    async def _start_tg(tg: TelegramChannel, name: str) -> None:
-        agent.channels[name] = tg
-        log.info("Starting Telegram bot (%s)…", name)
-        await tg.app.initialize()
-        await tg.app.start()
-        if tg.app.updater is not None:
-            await tg.app.updater.start_polling()
-
-    tg_global = config.channels.telegram
-    if tg_global.enabled and tg_global.bot_token:
-        await _start_tg(TelegramChannel(tg_global, agent, voice=voice), "telegram")
-
-    # -- Per-persona Telegram bots (#29): each persona with its own token is its
-    # own contact. Channel name "telegram:<persona>" silos history and resolves
-    # straight to that persona. ACL inherits the global allowlist when unset.
-    for persona in await agent.personae.list_personae():
-        token = (persona.bot_token or "").strip()
-        if not token or token == tg_global.bot_token:
-            continue  # no token, or shares the default bot's token — skip
-        pconf = TelegramConfig(
-            enabled=True,
-            bot_token=token,
-            allowed_user_ids=persona.allowed_user_ids or tg_global.allowed_user_ids,
-            topics_enabled=tg_global.topics_enabled,
-        )
-        await _start_tg(
-            TelegramChannel(pconf, agent, voice=voice, channel_name=f"telegram:{persona.name}"),
-            f"telegram:{persona.name}",
-        )
+    # -- Telegram: the default bot plus one bot per persona that carries a token (#29).
+    # A single bad/revoked token must never abort the others, WhatsApp, or the
+    # scheduler — each bot is brought up independently and failures are isolated.
+    async def _start_tg(conf, name: str, channel_name: str = "telegram") -> None:
+        try:
+            tg = TelegramChannel(conf, agent, voice=voice, channel_name=channel_name)
+            await tg.app.initialize()
+            await tg.app.start()
+            if tg.app.updater is not None:
+                await tg.app.updater.start_polling()
+            agent.channels[name] = tg  # registered only once it is actually polling
+            log.info("Telegram bot started (%s)", name)
+        except Exception:
+            log.exception("Failed to start Telegram bot %s — skipping", name)
+
+    try:
+        tg_global = config.channels.telegram
+        seen_tokens: set[str] = set()
+        if tg_global.enabled and tg_global.bot_token:
+            seen_tokens.add(tg_global.bot_token)
+            await _start_tg(tg_global, "telegram")
+
+        for persona in await agent.personae.list_personae():
+            token = (persona.bot_token or "").strip()
+            if not token:
+                continue  # no own bot — reachable only via the default bot
+            if token in seen_tokens:
+                log.warning(
+                    "Persona %s shares a bot token with another bot — skipping its bot "
+                    "(one token can only be polled once)",
+                    persona.name,
+                )
+                continue
+            seen_tokens.add(token)
+            pconf = TelegramConfig(
+                enabled=True,
+                bot_token=token,
+                allowed_user_ids=persona.allowed_user_ids or tg_global.allowed_user_ids,
+                topics_enabled=tg_global.topics_enabled,
+            )
+            await _start_tg(pconf, f"telegram:{persona.name}", f"telegram:{persona.name}")
+
+        # -- WhatsApp --
+        if config.channels.whatsapp.enabled:
+            from core.wacli import WacliManager
+
+            wacli = WacliManager()
+            wa = WhatsAppChannel(config.channels.whatsapp, agent, wacli=wacli)
+            agent.channels["whatsapp"] = wa
+            log.info("WhatsApp channel enabled (wacli)")
+
+        # -- Scheduler --
+        await agent.scheduler.load_jobs()
+        agent.scheduler.start()
+        log.info("Scheduler started with %d jobs", len(agent.scheduler.scheduler.get_jobs()))
+    except Exception:
+        # Bring-up failed after some bots were already polling — stop them so we
+        # don't leak orphaned pollers (which would 409 on the next start).
+        await _stop_telegram_bots(agent)
+        raise
 
-    # -- WhatsApp --
-    if config.channels.whatsapp.enabled:
-        from core.wacli import WacliManager
+    return agent
 
-        wacli = WacliManager()
-        wa = WhatsAppChannel(config.channels.whatsapp, agent, wacli=wacli)
-        agent.channels["whatsapp"] = wa
-        log.info("WhatsApp channel enabled (wacli)")
 
-    # -- Scheduler --
-    await agent.scheduler.load_jobs()
-    agent.scheduler.start()
-    log.info("Scheduler started with %d jobs", len(agent.scheduler.scheduler.get_jobs()))
+async def _stop_telegram_bots(agent) -> None:
+    """Stop and deregister the default bot and every per-persona bot (#29).
 
-    return agent
+    Each bot is torn down independently: one that fails to stop must not strand
+    the rest still polling (which would 409 on the next start).
+    """
+    for name, ch in list(agent.channels.items()):
+        if name != "telegram" and not name.startswith("telegram:"):
+            continue
+        try:
+            if ch.app.updater is not None:
+                await ch.app.updater.stop()
+            await ch.app.stop()
+            await ch.app.shutdown()
+        except Exception:
+            log.exception("Error stopping Telegram bot %s", name)
+        agent.channels.pop(name, None)
 
 
 async def _stop_agent(agent) -> None:
@@ -140,12 +175,7 @@ async def _stop_agent(agent) -> None:
 
     set_agent_context(None)
 
-    # Stop the default bot and every per-persona bot ("telegram:<persona>", #29).
-    for name, ch in list(agent.channels.items()):
-        if name == "telegram" or name.startswith("telegram:"):
-            await ch.app.updater.stop()
-            await ch.app.stop()
-            await ch.app.shutdown()
+    await _stop_telegram_bots(agent)
 
 
 # ---------------------------------------------------------------------------

core/scheduler.py

@@ -64,12 +64,17 @@ async def run_agent_task(
         )
 
     log.info("Scheduler running agent task: %s", task[:100])
+    # A "telegram:<persona>" job is generated AS that persona (#29) so the bot
+    # that delivers it also writes it — while keeping the "system" execution mode
+    # (auto-approved writes, no memory/reflection). Bare channels keep the default.
+    gen_persona = channel.split(":", 1)[1] if channel.startswith("telegram:") else None
     try:
         response = await agent.process(
             message=task,
             channel="system",
             user_id="scheduler",
             chat_id="scheduler",
+            persona_name=gen_persona,
         )
 
         # Deliver the response to the target channel

tests/test_personae_admin.py

@@ -107,8 +107,9 @@ def test_persona_raw_markdown_upsert(tmp_path) -> None:
 
 
 def test_persona_bot_fields_persist(tmp_path) -> None:
-    # Per-persona Telegram bot (#29): token + ACL survive the round-trip and the
-    # ACL is parsed from a comma-separated string into ints.
+    # Per-persona Telegram bot (#29): token + ACL survive the round-trip; the ACL
+    # is parsed from a comma-separated string into ints; the token is REDACTED on
+    # read (a secret, like the global Telegram token) but stored in full.
     client, _ = _client(tmp_path)
     r = client.post(
         "/personae",
@@ -122,7 +123,11 @@ def test_persona_bot_fields_persist(tmp_path) -> None:
     )
     assert r.status_code == 200
     got = client.get("/personae/coach", headers=AUTH).json()
-    assert got["bot_token"] == "123456:ABC-DEF"
+    # Redacted on read, but head/tail prove the full value reached storage.
+    assert "***" in got["bot_token"]
+    assert got["bot_token"].startswith("1234") and got["bot_token"].endswith("-DEF")
+    assert got["bot_token"] != "123456:ABC-DEF"
+    assert "123456:ABC-DEF" not in got["markdown"]  # not leaked via the raw view either
     assert got["allowed_user_ids"] == [111, 222]
 
 

tests/test_scheduler.py

@@ -93,6 +93,31 @@ async def test_run_agent_task_sends_to_owner() -> None:
     channel.send.assert_awaited_once_with(123, "done")
 
 
+@pytest.mark.asyncio
+async def test_run_agent_task_persona_job_generates_as_persona() -> None:
+    # A "telegram:<persona>" job (#29) is generated AS that persona (persona_name
+    # forced) while keeping the "system" execution mode, and delivered via that
+    # bot to its own owner (the bot's allowlist, not the global one).
+    channel = AsyncMock()
+    channel.config = SimpleNamespace(allowed_user_ids=[99])
+    agent = SimpleNamespace(
+        channels={"telegram:coach": channel},
+        process=AsyncMock(return_value=SimpleNamespace(text="done")),
+        config=SimpleNamespace(
+            channels=SimpleNamespace(telegram=SimpleNamespace(allowed_user_ids=[1]))
+        ),
+        job_store=None,
+    )
+    set_agent_context(agent)
+
+    await run_agent_task("ping", channel="telegram:coach")
+
+    _, kwargs = agent.process.call_args
+    assert kwargs["persona_name"] == "coach"
+    assert kwargs["channel"] == "system"  # execution mode unchanged
+    channel.send.assert_awaited_once_with(99, "done")  # coach bot → coach's owner
+
+
 @pytest.mark.asyncio
 async def test_run_agent_task_marks_oneshot_done() -> None:
     """One-shot jobs should be marked 'done' after execution."""