fix(agent): side-question review fixes (unattended turns, profile-bound leases, billing, attribution)

- No side_question_token for cron runs (platform="cron") or delegated subagent turns
  (_delegate_depth > 0): nobody reads the answer, so it is pure unattended spend.
- A lease is bound to the profile home it was issued in; a plugin context of another profile
  gets None and leaves the lease for its owner.
- Usage bills to the session the token was issued for (capture.session_id), not the parent's
  id at release time, which can rotate between issue and redeem.
- Plugin side questions report as aux_task="plugin_side_question" with the asking plugin_id in
  pre/post_auxiliary_call, and record usage as plugin_side_question:<plugin_id>; /btw keeps
  side_question.
- Drop the new legacy key llm.allow_side_question: the capability is declare + consent only.
- The timeout starts before the fork build, so runtime resolution counts against it.

agent/auxiliary_hooks.py

@@ -113,6 +113,8 @@ class _AuxCallHooks:
             started_at=self.started_at,
             message_count=len(kwargs.get("messages") or kwargs.get("input") or []),
         )
+        if metadata.get("plugin_id"):  # a call a plugin asked for (ctx.side_question)
+            self.base["plugin_id"] = str(metadata["plugin_id"])
 
     def pre(self) -> None:
         if not _has_hook(PRE_AUXILIARY_CALL):

agent/background_review.py

@@ -776,14 +776,16 @@ def _snapshot_review_usage(review_agent: Any) -> dict[str, Any]:
 
 def _record_review_usage_to_parent(
     parent_agent: Any, usage: dict[str, Any], *, task: str = "background_review",
+    session_id: Optional[str] = None,
 ) -> None:
-    """Record a fork's usage against the parent session under ``task`` (best-effort, never
-    raises). The fork has ``_session_db = None`` so conversation_loop's DB-gated accounting never
-    sees its calls; route them through the aux-accounting chokepoint, which writes only
-    ``session_model_usage`` — never the transcript or ``sessions`` row."""
+    """Record a fork's usage against the parent session (or ``session_id`` when given) under
+    ``task`` (best-effort, never raises). The fork has ``_session_db = None`` so
+    conversation_loop's DB-gated accounting never sees its calls; route them through the
+    aux-accounting chokepoint, which writes only ``session_model_usage`` — never the transcript
+    or ``sessions`` row."""
     try:
         session_db = getattr(parent_agent, "_session_db", None)
-        session_id = getattr(parent_agent, "session_id", None)
+        session_id = session_id or getattr(parent_agent, "session_id", None)
         counts = {key: int(usage.get(key) or 0) for key in _USAGE_COUNTERS}
         if session_db is None or not session_id or not any(counts.values()):
             return  # no DB, or the fork made no successful API calls (e.g. failed at spawn)
@@ -1146,13 +1148,16 @@ def _release_fork_clients(review_agent: Any) -> None:
         review_agent.release_clients()
 
 
-def release_cache_parity_fork(parent_agent: Any, fork: Any, *, task: str) -> dict[str, Any]:
+def release_cache_parity_fork(
+    parent_agent: Any, fork: Any, *, task: str, session_id: Optional[str] = None,
+) -> dict[str, Any]:
     """Teardown for a one-shot :func:`build_cache_parity_fork` caller (``/btw``, plugin side
-    questions): attribute the fork's usage to the parent session under ``task``, then release
-    only its clients (never ``close()``, which tears down the shared session's terminal sandbox,
-    browser and file state). Never raises; returns the usage snapshot."""
+    questions): attribute the fork's usage under ``task`` to ``session_id`` (default: the parent's
+    current session), then release only its clients (never ``close()``, which tears down the
+    shared session's terminal sandbox, browser and file state). Never raises; returns the usage
+    snapshot."""
     usage = _snapshot_review_usage(fork)
-    _record_review_usage_to_parent(parent_agent, usage, task=task)
+    _record_review_usage_to_parent(parent_agent, usage, task=task, session_id=session_id)
     _release_fork_clients(fork)
     return usage
 

agent/side_question_lease.py

@@ -5,7 +5,9 @@ thread captures the request inputs the side call will replay and hands ``post_ll
 opaque token. A token buys ONE side request shared by every plugin that receives the hook:
 redemption pops the lease whether it succeeds or not. It expires after ``LEASE_TTL_SECONDS`` or
 as soon as the parent agent starts its next turn. The captured ``contextvars`` context binds the
-owning profile and secret scope on whatever thread redeems the token.
+owning profile and secret scope on whatever thread redeems the token, and only a plugin context
+of that same profile can redeem it. Unattended turns (cron jobs, delegated subagents) get no
+token: nobody is there to read an answer bought with the main model's tokens.
 """
 
 from __future__ import annotations
@@ -35,6 +37,7 @@ MAX_TIMEOUT_SECONDS = 60.0
 class _Lease:
     capture: SideQuestionCapture
     expires_at: float
+    home_key: str  # hermes_home_key() of the profile the turn ran under
 
 
 _leases: dict[str, _Lease] = {}
@@ -61,11 +64,19 @@ def _purge_locked(now: float, agent: Any = None) -> None:
         del _leases[stale]
 
 
+def _unattended(agent: Any) -> bool:
+    """A turn nobody is waiting on: a persistence-detached fork, a cron run, a delegated child."""
+    return (getattr(agent, "_persist_disabled", False) or getattr(agent, "platform", None) == "cron"
+            or getattr(agent, "_delegate_depth", 0) > 0)
+
+
 def issue_side_question_token(agent: Any, messages: list[dict[str, Any]], turn_id: str) -> Optional[str]:
-    """Capture the finished turn and lease it under a fresh token; None (nothing captured) for a
-    persistence-detached fork or when no consented plugin is loaded. Turn thread only: the
-    captured context is the turn's."""
-    if getattr(agent, "_persist_disabled", False) or not _consented_plugin_loaded():
+    """Capture the finished turn and lease it under a fresh token; None (nothing captured) for an
+    unattended turn (fork, cron run, subagent) or when no consented plugin is loaded. Turn thread
+    only: the captured context and profile are the turn's."""
+    from hermes_constants import hermes_home_key
+
+    if _unattended(agent) or not _consented_plugin_loaded():
         return None
     capture = SideQuestionCapture(
         parent_ref=weakref.ref(agent),
@@ -84,7 +95,7 @@ def issue_side_question_token(agent: Any, messages: list[dict[str, Any]], turn_i
         # Drop this agent's older leases (a new turn) and every expired or orphaned one (each holds
         # a transcript), so the table holds at most one lease per live session agent.
         _purge_locked(now, agent)
-        _leases[token] = _Lease(capture, now + LEASE_TTL_SECONDS)
+        _leases[token] = _Lease(capture, now + LEASE_TTL_SECONDS, hermes_home_key())
     return token
 
 
@@ -105,13 +116,23 @@ def _redeemable(lease: Optional[_Lease], prompt: Any) -> Optional[str]:
 
 
 def redeem_side_question(
-    token: Any, prompt: Any, *, plugin_id: str, timeout: float = DEFAULT_TIMEOUT_SECONDS,
+    token: Any, prompt: Any, *, plugin_id: str, home: Any = None, timeout: float = DEFAULT_TIMEOUT_SECONDS,
 ) -> Optional[str]:
     """Spend the lease behind ``token`` on one text-only question; the answer or None.
 
-    The lease is consumed by this call whatever the outcome."""
+    ``home`` is the redeeming plugin's profile home (the active one when omitted). A lease from
+    another profile is refused and left for its owner; otherwise this call consumes it whatever
+    the outcome."""
+    from hermes_constants import hermes_home_key
+
+    home_key = hermes_home_key(home)
     with _lock:
-        lease = _leases.pop(token, None) if isinstance(token, str) else None
+        lease = _leases.get(token) if isinstance(token, str) else None
+        if lease is not None and lease.home_key != home_key:
+            logger.info("side_question plugin=%s outcome=rejected:other_profile", plugin_id)
+            return None
+        if lease is not None:
+            del _leases[token]
         _purge_locked(time.monotonic())
     reason = _redeemable(lease, prompt)
     if reason is not None:

agent/side_question_request.py

@@ -6,9 +6,10 @@ messages) verbatim plus one user prompt, so the request reads the parent's warm
 It runs the turn loop's own request phases (assembly, the payload build with its LLM
 middleware, ``perform_api_call``) exactly once: no tool dispatch (tool calls in the response
 are dropped), no retries, no fallback model, no compression, no persistence. Like any other
-side call it reports through ``pre/post_auxiliary_call`` (``aux_task="side_question"``), never
-the turn-scoped ``pre/post_api_request``, and its usage is attributed to the parent session as
-``side_question``.
+side call it reports through ``pre/post_auxiliary_call`` (``aux_task="plugin_side_question"``
+plus the asking ``plugin_id``), never the turn-scoped ``pre/post_api_request``, and its usage is
+billed to the session the token was issued for as ``plugin_side_question:<plugin_id>`` (``/btw``
+keeps ``side_question``).
 """
 
 from __future__ import annotations
@@ -24,6 +25,8 @@ from typing import Any, Optional
 logger = logging.getLogger(__name__)
 
 SIDE_QUESTION_ORIGIN = "side_question"
+# Hook/usage task for plugin side questions, distinct from /btw's ``side_question``.
+PLUGIN_SIDE_QUESTION_TASK = "plugin_side_question"
 
 # Tool entries the client executes; any other ``type`` (Responses ``web_search``, Anthropic
 # ``web_search_20250305``, ...) runs server-side and could act without Hermes seeing it.
@@ -160,7 +163,8 @@ def _send(fork: Any, capture: SideQuestionCapture, prompt: str, audit: _Audit) -
         audit.outcome = "refused:server_side_tools"
         return None
     hooks = start_aux_call_hooks(
-        aux_task=SIDE_QUESTION_ORIGIN, metadata={"api_request_id": api_request_id}, client=fork.client,
+        aux_task=PLUGIN_SIDE_QUESTION_TASK, metadata={"api_request_id": api_request_id, "plugin_id": audit.plugin_id},
+        client=fork.client,
         kwargs=built.api_kwargs, provider=str(fork.provider or ""), model=str(fork.model or ""),
         api_mode=str(fork.api_mode or ""), streaming=False,
     )
@@ -237,23 +241,31 @@ def ask_text_only(
         _log_audit(audit)
         return None
     fork = None
+    built: list = []  # the fork once built, for the timer
     timed_out = threading.Event()
-    timer = None
+
+    def _on_timeout() -> None:
+        timed_out.set()
+        # Soft interrupt with a system reason: aborts the in-flight request without being
+        # recorded as a user Stop.
+        for target in built:
+            target.interrupt(None, tool_reason="side question timed out")
+
+    # The budget starts before the build: runtime resolution (credentials, client setup) is part
+    # of the caller's wait. A build cannot be interrupted, so one that outlasts the budget is
+    # caught right after it and sends nothing.
+    timer = threading.Timer(timeout, _on_timeout)
+    timer.daemon = True
+    timer.start()
     try:
         fork = _build_fork(parent, capture)
+        built.append(fork)
+        if timed_out.is_set():
+            audit.outcome = "timeout"
+            return None
         if runtime_key(parent) != capture.runtime:
             audit.outcome = "refused:runtime_changed"
             return None
-
-        def _on_timeout() -> None:
-            timed_out.set()
-            # Soft interrupt with a system reason: aborts the in-flight request without being
-            # recorded as a user Stop.
-            fork.interrupt(None, tool_reason="side question timed out")
-
-        timer = threading.Timer(timeout, _on_timeout)
-        timer.daemon = True
-        timer.start()
         assistant_message = _send(fork, capture, prompt, audit)
         if assistant_message is None:
             return None
@@ -266,8 +278,11 @@ def ask_text_only(
         logger.warning("side question %s (plugin=%s)", audit.outcome, plugin_id or "?", exc_info=True)
         return None
     finally:
-        if timer is not None:
-            timer.cancel()
+        timer.cancel()
         if fork is not None:
-            audit.usage = release_cache_parity_fork(parent, fork, task=SIDE_QUESTION_ORIGIN)
+            # Billed to the session the token was issued for: the parent's id may have rotated
+            # (``/compress``, a session reset) since the turn ended.
+            audit.usage = release_cache_parity_fork(
+                parent, fork, task=f"{PLUGIN_SIDE_QUESTION_TASK}:{plugin_id or '?'}",
+                session_id=capture.session_id)
         _log_audit(audit)

hermes_cli/plugin_capabilities.py

@@ -20,7 +20,9 @@ class CapabilitySpec:
     """One declarable capability and the legacy gate it maps to."""
 
     id: str
-    legacy_path: tuple[str, ...]  # deprecated boolean under plugins.entries.<id>, e.g. ("llm", "allow_model_override")
+    # Deprecated boolean under plugins.entries.<id>, e.g. ("llm", "allow_model_override"); empty for a
+    # capability born after the consent flow, whose only gate is declare + consent.
+    legacy_path: tuple[str, ...]
     description: str  # one-line risk description shown on the consent screen
 
 
@@ -42,7 +44,7 @@ _CAPABILITY_ROWS = (
     ("llm.task_override", ("llm", "allow_task_override"),
      "Route its LLM calls through the host's built-in auxiliary "
      "task lanes"),
-    ("llm.side_question", ("llm", "allow_side_question"),
+    ("llm.side_question", (),
      "Ask your agent a short question after each turn. It sees the "
      "whole session, including memory, and spends your main model's tokens."),
     ("gateway.platform_actions", ("allow_platform_actions",),
@@ -121,7 +123,9 @@ def granted_capabilities(plugin_id: str, config: Optional[Mapping[str, Any]] = N
 
 
 def _legacy_gate_set(entry: Mapping[str, Any], spec: CapabilitySpec) -> bool:
-    """True when the deprecated ``allow_*`` key for *spec* is truthy."""
+    """True when the deprecated ``allow_*`` key for *spec* is truthy (never for a consent-only one)."""
+    if not spec.legacy_path:
+        return False
     node: Any = entry
     for part in spec.legacy_path:
         if not isinstance(node, Mapping):
@@ -177,6 +181,8 @@ def record_consent(plugin_id: str, granted: Iterable[str], declared: Iterable[st
     }
     # Bridge: mirror each grant into its legacy gate (enforcement sites still read allow_*).
     for cap in entry[GRANTED_KEY]:
+        if not CAPABILITY_REGISTRY[cap].legacy_path:
+            continue
         *parents, leaf = CAPABILITY_REGISTRY[cap].legacy_path
         node = entry
         for part in parents:

hermes_cli/plugins_capability_ctx.py

@@ -28,7 +28,8 @@ class PluginCapabilityCtxMixin:
         """Ask the session's own agent ``prompt`` about the turn that just ended; text or None.
 
         ``token`` is ``post_llm_call``'s ``side_question_token``: one attempt per turn, shared by
-        every plugin, valid for 60 s and only until the next turn starts. Same model, the whole
+        every plugin of the session's profile, valid for 60 s and only until the next turn starts
+        (cron runs and subagent turns get no token). Same model, the whole
         session (memory included) and its warm prompt cache; tools never run and nothing is
         written to the transcript. Needs the ``llm.side_question`` capability; without it this
         returns None and leaves the attempt to a consented plugin."""
@@ -40,7 +41,8 @@ class PluginCapabilityCtxMixin:
                 logger.info("Plugin %s called ctx.side_question without the %s capability; returning None",
                             self.plugin_id, SIDE_QUESTION_CAPABILITY)
             return None
-        return redeem_side_question(token, prompt, plugin_id=self.plugin_id, timeout=timeout)
+        return redeem_side_question(
+            token, prompt, plugin_id=self.plugin_id, home=self._manager.home_path, timeout=timeout)
 
     def _side_question_granted(self, capability: str) -> bool:
         """Consent read from THIS manager's profile home: plugins call from their own threads,

tests/agent/test_side_question_lease.py

@@ -109,6 +109,18 @@ def test_no_token_for_a_persist_disabled_fork(executor, consented):
     assert issue_side_question_token(_agent(_persist_disabled=True), [], "turn-1") is None
 
 
+def test_no_token_for_a_cron_run(executor, consented):
+    """A scheduled job has nobody watching: a plugin must not spend the main model on it."""
+    assert issue_side_question_token(_agent(platform="cron"), [], "turn-1") is None
+    assert isinstance(issue_side_question_token(_agent(platform="telegram"), [], "turn-1"), str)
+
+
+def test_no_token_for_a_subagent_turn(executor, consented):
+    """A delegated child's turn is unattended work inside the parent's turn."""
+    assert issue_side_question_token(_agent(_delegate_depth=1), [], "turn-1") is None
+    assert isinstance(issue_side_question_token(_agent(_delegate_depth=0), [], "turn-1"), str)
+
+
 def test_capture_is_frozen_at_issue(executor, consented):
     messages = [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}]
     agent = _agent()  # the lease holds its parent weakly
@@ -197,6 +209,25 @@ def test_bad_prompt_is_none_and_consumes_the_lease(executor, consented, prompt):
     assert executor == []
 
 
+def test_another_profile_cannot_redeem_the_lease(executor, consented, tmp_path):
+    """A lease is bound to the profile home it was issued in: a plugin context from another
+    profile's manager (granted there too) gets None and leaves the lease to its owner."""
+    other_home = tmp_path / "profiles" / "other"
+    other_home.mkdir(parents=True)
+    (other_home / "config.yaml").write_text(yaml.safe_dump(
+        {"plugins": {"enabled": ["next-prompt"], "entries": {"next-prompt": {"granted_capabilities": [CAP]}}}}),
+        encoding="utf-8")
+    foreign = PluginContext(consented._plugins["next-prompt"].manifest,
+                            plugins_mod.PluginManager(scope_key=str(other_home)))
+    agent = _agent()  # the lease holds its parent weakly
+    token = issue_side_question_token(agent, [], "turn-1")
+
+    assert foreign.side_question(token, "q") is None
+    assert redeem_side_question(token, "q", plugin_id="next-prompt", home=other_home) is None
+    assert executor == []
+    assert _ctx(consented, "next-prompt").side_question(token, "q") == "answer for next-prompt"
+
+
 @pytest.mark.parametrize("given, used", [(0.5, 0.5), (600, 60.0), (0, 15.0), (-3, 15.0), (float("nan"), 15.0), ("x", 15.0)])
 def test_timeout_is_clamped(executor, consented, given, used):
     agent = _agent()  # the lease holds its parent weakly
@@ -260,7 +291,7 @@ def _observed_turn_with_side_question(monkeypatch, side_reply) -> list[dict]:
         f"EVENTS = pathlib.Path({str(events)!r})\n\n"
         "def _log(name, **kw):\n"
         "    row = {'event': name, 'aux_task': kw.get('aux_task'), 'usage': kw.get('usage'),\n"
-        "           'error_type': kw.get('error_type')}\n"
+        "           'error_type': kw.get('error_type'), 'plugin_id': kw.get('plugin_id')}\n"
         "    with EVENTS.open('a') as fh:\n"
         "        fh.write(json.dumps(row, default=str) + '\\n')\n\n"
         "def register(ctx):\n"
@@ -300,14 +331,17 @@ def _observed_turn_with_side_question(monkeypatch, side_reply) -> list[dict]:
 
 def test_side_request_fires_auxiliary_hooks_not_turn_request_hooks(monkeypatch):
     """Turn-scoped observers (``pre/post_api_request``) see only the turn's own request; the
-    side request reports through ``pre/post_auxiliary_call`` with ``aux_task='side_question'``."""
+    side request reports through ``pre/post_auxiliary_call`` as ``plugin_side_question`` (not
+    ``/btw``'s ``side_question``), naming the plugin that asked."""
     usage = SimpleNamespace(prompt_tokens=900, completion_tokens=7, total_tokens=907)
     rows = _observed_turn_with_side_question(monkeypatch, SimpleNamespace(
         choices=[SimpleNamespace(message=SimpleNamespace(content="Ask why it is 42.", tool_calls=None),
                                  finish_reason="stop")], model="test/model", usage=usage))
     assert [r["event"] for r in rows] == [
         "pre_api_request", "post_api_request", "pre_auxiliary_call", "post_auxiliary_call"]
-    assert rows[2]["aux_task"] == rows[3]["aux_task"] == "side_question"
+    assert rows[2]["aux_task"] == rows[3]["aux_task"] == "plugin_side_question"
+    assert rows[2]["plugin_id"] == rows[3]["plugin_id"] == "observer"
+    assert rows[0]["plugin_id"] is None  # the turn's own request names no plugin
     assert rows[3]["usage"]["output_tokens"] == 7 and rows[3]["error_type"] is None
 
 

tests/agent/test_side_question_request.py

@@ -24,6 +24,8 @@ from hermes_state import SessionDB
 from run_agent import AIAgent
 
 SIDE_PROMPT = "Suggest the user's next prompt in one line."
+# Plugin side questions bill per plugin, apart from ``/btw``'s ``side_question``.
+PLUGIN_TASK = "plugin_side_question:next-prompt"
 
 
 def _tool_defs(*names: str) -> list:
@@ -263,7 +265,7 @@ def test_side_call_writes_only_one_usage_row(provider, issued, tmp_path):
     parent = _make_parent(session_db=db, session_id="sess-side")
     _run_parent_turn(parent, provider)
     rows_before = _message_rows(db, "sess-side")
-    usage_before = [r for r in _usage_rows(db, "sess-side") if r["task"] == "side_question"]
+    usage_before = [r for r in _usage_rows(db, "sess-side") if r["task"] == PLUGIN_TASK]
     assert rows_before and usage_before == []
     memory_before = {p.name: p.read_bytes() for p in memories.iterdir()}
 
@@ -271,11 +273,26 @@ def test_side_call_writes_only_one_usage_row(provider, issued, tmp_path):
     assert redeem_side_question(issued[-1], SIDE_PROMPT, plugin_id="next-prompt") == "Ask: why 42?"
 
     assert _message_rows(db, "sess-side") == rows_before
-    side_rows = [r for r in _usage_rows(db, "sess-side") if r["task"] == "side_question"]
+    side_rows = [r for r in _usage_rows(db, "sess-side") if r["task"] == PLUGIN_TASK]
     assert len(side_rows) == 1 and side_rows[0]["api_call_count"] == 1
     assert {p.name: p.read_bytes() for p in memories.iterdir()} == memory_before
 
 
+def test_side_call_bills_the_session_the_token_was_issued_for(provider, issued, tmp_path):
+    """The parent's session id can rotate between issue and redeem (``/compress``, a session
+    reset): the side call's spend belongs to the session whose turn it replays."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    parent = _make_parent(session_db=db, session_id="sess-issued")
+    _run_parent_turn(parent, provider, tool_round=False)
+    parent.session_id = "sess-rotated"
+
+    provider.replies.append(_response("Ask: why 42?", usage=_usage()))
+    assert redeem_side_question(issued[-1], SIDE_PROMPT, plugin_id="next-prompt") == "Ask: why 42?"
+
+    assert [r["api_call_count"] for r in _usage_rows(db, "sess-issued") if r["task"] == PLUGIN_TASK] == [1]
+    assert _usage_rows(db, "sess-rotated") == []
+
+
 def test_side_call_leaves_the_parent_sessions_terminal_environment_alive(provider, issued):
     """The fork shares the parent's session_id, and CLI/gateway key a session's terminal
     sandbox by it: tearing the fork down must not tear down the session's live tool state."""
@@ -301,8 +318,8 @@ def test_side_call_leaves_the_parent_sessions_terminal_environment_alive(provide
 
 
 def test_side_call_runs_under_the_turns_profile_and_secret_scope(provider, issued, tmp_path):
-    """The plugin redeems from its own thread, where no profile is bound; the request still
-    resolves the parent turn's HERMES_HOME and secret scope."""
+    """The plugin (of the turn's profile) redeems from its own thread, where no profile is bound;
+    the request still resolves the parent turn's HERMES_HOME and secret scope."""
     from agent.secret_scope import current_secret_scope_home, reset_secret_scope, set_secret_scope
     from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override
 
@@ -326,7 +343,8 @@ def test_side_call_runs_under_the_turns_profile_and_secret_scope(provider, issue
     provider.replies.append(_reply)
     answers: list = []
     worker = threading.Thread(
-        target=lambda: answers.append(redeem_side_question(issued[-1], SIDE_PROMPT, plugin_id="next-prompt")))
+        target=lambda: answers.append(
+            redeem_side_question(issued[-1], SIDE_PROMPT, plugin_id="next-prompt", home=profile_home)))
     worker.start()
     worker.join(10)
     assert answers == ["Ask: why 42?"]
@@ -446,6 +464,30 @@ def test_timeout_interrupts_the_inflight_request(provider, issued):
     assert interrupted == ["side question timed out"]
 
 
+def test_timeout_counts_the_fork_build(provider, issued, monkeypatch):
+    """Runtime resolution in the fork build is part of the caller's budget: a build that
+    outlasts the timeout sends nothing."""
+    from agent import side_question_request
+
+    parent = _make_parent()
+    _run_parent_turn(parent, provider, tool_round=False)
+    real_build = side_question_request._build_fork
+
+    def _slow_build(*args, **kwargs):
+        # Outlast the budget: hold the build until the side question's timer has fired (a Timer
+        # thread that is already gone has fired too).
+        for timer in [t for t in threading.enumerate() if isinstance(t, threading.Timer)
+                      and getattr(t.function, "__name__", "") == "_on_timeout"]:
+            assert timer.finished.wait(10)
+        return real_build(*args, **kwargs)
+
+    monkeypatch.setattr(side_question_request, "_build_fork", _slow_build)
+    provider.replies.append(_response("too late", usage=_usage()))
+    sent = len(provider.requests)
+    assert redeem_side_question(issued[-1], SIDE_PROMPT, plugin_id="next-prompt", timeout=0.1) is None
+    assert len(provider.requests) == sent
+
+
 def test_provider_error_returns_none_without_fallback(provider, issued):
     parent = _make_parent(fallback_model={"provider": "openrouter", "model": "fallback/model"})
     _run_parent_turn(parent, provider, tool_round=False)

tests/hermes_cli/test_plugin_capabilities.py

@@ -43,8 +43,9 @@ def _read_cfg(home):
 
 class TestRegistry:
     def test_every_capability_has_legacy_gate(self):
+        # llm.side_question postdates the consent flow: declare + consent is its only gate.
         for spec in CAPABILITY_REGISTRY.values():
-            assert spec.legacy_path, spec.id
+            assert bool(spec.legacy_path) is (spec.id != "llm.side_question"), spec.id
             assert spec.description
 
 
@@ -296,6 +297,21 @@ class TestLegacyGateCompat:
         )
         assert plugin_capability_granted("oldplug", "llm.model_override") is True
 
+    def test_consent_only_capability_ignores_allow_keys(self, hermes_home):
+        """``llm.side_question`` has no legacy key: an ``allow_*`` lookalike (or any other
+        entry content) never opens it; only a recorded grant does."""
+        (hermes_home / "config.yaml").write_text(
+            "plugins:\n  entries:\n    newplug:\n"
+            "      llm:\n        allow_side_question: true\n",
+            encoding="utf-8",
+        )
+        assert plugin_capability_granted("newplug", "llm.side_question") is False
+        record_consent("freshplug", ["llm.side_question"], ["llm.side_question"])
+        entry = _read_cfg(hermes_home)["plugins"]["entries"]["freshplug"]
+        assert entry["granted_capabilities"] == ["llm.side_question"]
+        assert "llm" not in entry  # no legacy key to mirror the grant into
+        assert plugin_capability_granted("freshplug", "llm.side_question") is True
+
     def test_legacy_false_stays_denied(self, hermes_home):
         (hermes_home / "config.yaml").write_text(
             "plugins:\n  entries:\n    oldplug:\n"

website/docs/developer-guide/plugin-llm-access.md

@@ -568,15 +568,20 @@ capabilities:
 ### Lease rules
 
 * `post_llm_call` receives `side_question_token` (a string) when at
-  least one loaded plugin holds the capability, else `None`.
+  least one loaded plugin holds the capability, else `None`. Cron runs
+  and delegated subagent turns always get `None`: nobody is waiting on
+  them, so a side question would only spend tokens.
 * **One attempt per turn, shared by every plugin.** Every plugin's
   hook gets the same token; the first `ctx.side_question` call spends
   it, whether it succeeds or not. A plugin without the capability gets
   `None` back and does not spend it.
 * The token expires after **60 seconds**, and as soon as the session
   starts its next turn.
+* The token belongs to the profile whose turn issued it; a plugin
+  context of another profile gets `None` and does not spend it.
 * `prompt` is a non-empty string of at most **4096 characters**.
-* `timeout` defaults to **15 seconds** and is capped at 60.
+* `timeout` defaults to **15 seconds** and is capped at 60. It covers
+  the whole call, including setting up the request.
 
 ### What you get back
 
@@ -591,13 +596,16 @@ and no fallback model.
 
 The call never runs a tool (tool calls in the reply are dropped) and
 never writes to the transcript. Its tokens are billed to the session
-and recorded in its usage as `side_question`. You can't pick the
+whose turn issued the token (even if the session id has since rotated)
+and recorded in its usage as `plugin_side_question:<plugin id>`, apart
+from `/btw`'s `side_question`. You can't pick the
 model, tools, system prompt or output length: changing any of them
 would miss the warm cache. Ask for brevity in the prompt.
 
 Observability plugins see the call as an auxiliary call:
 `pre_auxiliary_call` / `post_auxiliary_call` fire with
-`aux_task="side_question"`, and the turn-scoped `pre_api_request` /
+`aux_task="plugin_side_question"` and `plugin_id` set to the asking
+plugin, and the turn-scoped `pre_api_request` /
 `post_api_request` do not fire for it.
 
 ## Where this fits in the plugin surface

website/docs/user-guide/features/hooks.md

@@ -778,7 +778,7 @@ def my_callback(session_id: str, user_message: str, assistant_response: str,
 | `conversation_history` | `list` | Copy of the full message list after the turn completed |
 | `model` | `str` | The model identifier |
 | `platform` | `str` | Where the session is running |
-| `side_question_token` | `str \| None` | One-use token for [`ctx.side_question`](../../developer-guide/plugin-llm-access.md#asking-the-sessions-own-agent-ctxside_question), shared by every plugin; `None` unless a loaded plugin holds the `llm.side_question` capability |
+| `side_question_token` | `str \| None` | One-use token for [`ctx.side_question`](../../developer-guide/plugin-llm-access.md#asking-the-sessions-own-agent-ctxside_question), shared by every plugin of the session's profile; `None` unless a loaded plugin holds the `llm.side_question` capability, and always `None` for cron runs and subagent turns |
 
 **Fires:** In `agent/turn_finalizer.py` (`finalize_turn()`, called by `run_conversation()` in `agent/conversation_loop.py`), after the tool loop exits with a final response. Guarded by `if final_response and not interrupted` — so it does **not** fire when the user interrupts mid-turn or the agent hits the iteration limit without producing a response.
 
@@ -1814,7 +1814,7 @@ Fires for a failed provider attempt with status/retry timing, an `error` object,
 
 #### `pre_auxiliary_call` / `post_auxiliary_call`
 
-Auxiliary LLM calls — session titling, context compression, MoA advisors and the aggregator, vision, approval classification, memory and other side tasks — run outside the main tool-calling loop and do **not** fire `pre_api_request` / `post_api_request` (those stay turn-scoped, so a trace-per-turn plugin never sees side traffic by accident). Subscribe to `pre_auxiliary_call` / `post_auxiliary_call` instead: they fire once per physical provider attempt (retries and fallbacks included) with the same payload shape plus `aux_task` (the task name, e.g. `title_generation`, `compression`, `moa_aggregator`, `vision`, `side_question` for a plugin's [`ctx.side_question`](../../developer-guide/plugin-llm-access.md#asking-the-sessions-own-agent-ctxside_question)). `session_id` / `task_id` / `turn_id` are the parent turn's when the call runs under one, empty otherwise; `api_request_id` (`aux-…`) is shared by every attempt of one logical call and `retry_count` distinguishes them. Both are observer-only and fail-open: a raising or timed-out callback is logged and the auxiliary task proceeds. `post_auxiliary_call` carries `error` / `error_type` when the attempt raised and `streaming: True` (with `usage`/`response` `None`) when the response is handed back as a stream.
+Auxiliary LLM calls — session titling, context compression, MoA advisors and the aggregator, vision, approval classification, memory and other side tasks — run outside the main tool-calling loop and do **not** fire `pre_api_request` / `post_api_request` (those stay turn-scoped, so a trace-per-turn plugin never sees side traffic by accident). Subscribe to `pre_auxiliary_call` / `post_auxiliary_call` instead: they fire once per physical provider attempt (retries and fallbacks included) with the same payload shape plus `aux_task` (the task name, e.g. `title_generation`, `compression`, `moa_aggregator`, `vision`, `plugin_side_question` for a plugin's [`ctx.side_question`](../../developer-guide/plugin-llm-access.md#asking-the-sessions-own-agent-ctxside_question), which also carries the asking plugin's `plugin_id`). `session_id` / `task_id` / `turn_id` are the parent turn's when the call runs under one, empty otherwise; `api_request_id` (`aux-…`) is shared by every attempt of one logical call and `retry_count` distinguishes them. Both are observer-only and fail-open: a raising or timed-out callback is logged and the auxiliary task proceeds. `post_auxiliary_call` carries `error` / `error_type` when the attempt raised and `streaming: True` (with `usage`/`response` `None`) when the response is handed back as a stream.
 
 ### `on_skill_lifecycle`
 

website/docs/user-guide/features/plugins.md

@@ -603,11 +603,12 @@ working but are **deprecated** in favor of the consent flow:
 | `llm.agent_id_override` | `llm.allow_agent_id_override` |
 | `llm.profile_override` | `llm.allow_profile_override` |
 | `llm.task_override` | `llm.allow_task_override` |
-| `llm.side_question` | `llm.allow_side_question` |
 | `gateway.platform_actions` | `allow_platform_actions` |
 
 A gate is open when *either* the capability is granted *or* the legacy key is
 set — existing configs keep working unchanged.
+`llm.side_question` has no legacy key: it is open only when declared and
+granted through the consent flow.
 
 :::warning Not a sandbox
 Capabilities are a **consent and audit layer**, not isolation. Plugins run as