[F031] identity: role_for_slug_or_none so defensive skip-guards don't crash the dispatcher tick on stale slugs

roboco/foundation/identity.py

@@ -298,6 +298,19 @@ def role_for_slug(slug: str) -> Role:
     return agent_for_slug(slug).role
 
 
+def role_for_slug_or_none(slug: str) -> Role | None:
+    """Safe variant of :func:`role_for_slug` for defensive skip-guards.
+
+    Returns ``None`` for an unknown/stale slug instead of raising ``KeyError``,
+    so a stale assignee or notification-target slug can't crash the whole
+    dispatcher tick. Callers that only need to check "is this a human-only
+    role?" treat ``None`` as "not human-only" (``None in (CEO, ...)`` is False)
+    and proceed without raising.
+    """
+    row = AGENTS.get(slug)
+    return row.role if row is not None else None
+
+
 def team_for_slug(slug: str) -> Team:
     """Shorthand for `agent_for_slug(slug).team`."""
     return agent_for_slug(slug).team

roboco/runtime/orchestrator.py

@@ -46,7 +46,7 @@ from roboco.agents_config import (
 )
 from roboco.config import settings
 from roboco.foundation import identity as _foundation
-from roboco.foundation.identity import CELL_TEAMS, Role, role_for_slug
+from roboco.foundation.identity import CELL_TEAMS, Role, role_for_slug_or_none
 from roboco.foundation.policy.agent_loop import DEFAULT_BUDGET as _AGENT_LOOP_BUDGET
 from roboco.foundation.policy.batch import is_branchless_coordination
 from roboco.models import AgentRole, Team
@@ -1866,7 +1866,7 @@ class AgentOrchestrator:
         # here. Safe because the dedicated human-spawn paths do not route
         # through spawn_agent (see the _spawn_intake_container note at the top
         # of this file).
-        _role = role_for_slug(agent_id)
+        _role = role_for_slug_or_none(agent_id)
         if _role in (Role.CEO, Role.PROMPTER, Role.SECRETARY):
             logger.error(
                 "spawn_agent refused for human-only role — dispatchers must never"
@@ -9427,7 +9427,7 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
                 # Mirrors the spawn_agent human-role guard; a skip here keeps
                 # a mis-assigned human task from aborting this dispatcher's
                 # whole tick (the chokepoint would otherwise raise).
-                if role_for_slug(assigned_slug) in (
+                if role_for_slug_or_none(assigned_slug) in (
                     Role.CEO,
                     Role.PROMPTER,
                     Role.SECRETARY,
@@ -9604,7 +9604,11 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
         # act on through the panel; do NOT release it to pending (that would
         # re-route a human-owned task to a PM). See spawn_agent's human-role
         # guard for the structural backstop.
-        if role_for_slug(agent_slug) in (Role.CEO, Role.PROMPTER, Role.SECRETARY):
+        if role_for_slug_or_none(agent_slug) in (
+            Role.CEO,
+            Role.PROMPTER,
+            Role.SECRETARY,
+        ):
             return None
         # The assignee is running, and on THIS task — healthy.
         instance = self._instances.get(agent_slug)
@@ -10027,7 +10031,7 @@ Never `commit`, never write code, never run `git`. PMs coordinate.
                 # notification target (board-review handoff, escalation, etc.)
                 # is expected; it is NOT a spawn signal. Skip — the
                 # notification stays for the human to read in the panel.
-                if role_for_slug(agent_slug) in (
+                if role_for_slug_or_none(agent_slug) in (
                     Role.CEO,
                     Role.PROMPTER,
                     Role.SECRETARY,

tests/foundation/test_identity.py

@@ -229,3 +229,15 @@ def test_team_for_slug() -> None:
     assert identity.team_for_slug("be-dev-1") == identity.Team.BACKEND
     assert identity.team_for_slug("ceo") == identity.Team.BOARD
     assert identity.team_for_slug("head-marketing") == identity.Team.BOARD
+
+
+def test_role_for_slug_or_none_unknown_returns_none() -> None:
+    """F019/F031: a safe variant for defensive skip-guards — an unknown/stale
+    slug returns None instead of raising KeyError, so a stale assignee or
+    notification-target slug can't crash the whole dispatcher tick."""
+    assert identity.role_for_slug_or_none("nonexistent-slug") is None
+
+
+def test_role_for_slug_or_none_known_returns_role() -> None:
+    assert identity.role_for_slug_or_none("be-pm") == identity.Role.CELL_PM
+    assert identity.role_for_slug_or_none("ceo") == identity.Role.CEO