Spaces:

abc1181
/

Jaarvis

Starting

Jaarvis / core /agent_runtime.py
Wire bearer shim into delegation
a02eb85 verified 11 days ago
30.5 kB
"""The agent JARVIS delegates to — and the subagent layer underneath it.
Everything else in here is a tool JARVIS calls. This is the thing that *is* an
agent: one that plans, runs code, manages files, browses, and hands work to
subagents of its own.
Why the SDK rather than something hand-rolled
--------------------------------------------
`core/coder.py` shells out to a CLI and hopes. `core/agents.py` is a roster with
a budget. Neither can decide a task needs three steps in an unexpected order, and
neither can notice it is wrong and try again.
Google's Antigravity SDK is that harness already built, and — the part that
mattered when this was rewritten — it is not just a chat loop. `LocalAgentConfig`
carries, verified against the installed 3.x package rather than assumed:
    subagents                   list[SubagentConfig]   native subagents
    conversation_id             str                    a session that persists
    session_continuation_mode   RESUME | CREATE_OR_RESUME | CREATE_ONLY
    tools                       list[Callable | str]   per-agent allowlist
    policies                    list[Policy]           declarative policy
    hooks                       list[Inspect|DecideHook]
    budget_config               caps on tokens, model calls, tool calls
    workspaces                  list[str]              sandboxed dirs
    skills_paths, mcp_servers, compaction_config, retry_config, response_schema
and `BuiltinTools` includes `START_SUBAGENT`, so an agent can fan out on its own
rather than only when we ask it to.
The previous version of this file used four of those fields and ignored the rest.
It was one turn in, one answer out: no memory between calls, no subagents, no
per-role tool scoping, no spend ceiling, and nothing streamed while it worked.
That is not a delegation layer, it is a very slow function call.
What this module does with the SDK
----------------------------------
A hired agent is not a row in a JSON file. It is an Antigravity session with:
  * a **stable conversation_id**, so "Ada, pick up where you left off" works and
    so an agent accumulates context instead of starting from nothing each call;
  * a **scoped tool allowlist** derived from its role — a Researcher gets
    SEARCH_WEB and READ_URL_CONTENT and cannot reach RUN_COMMAND or EDIT_FILE,
    which is our approval model enforced by the runtime rather than by asking a
    model nicely;
  * a **BudgetConfig**, so a runaway loop hits a ceiling instead of a credit card;
  * its own **workspace** directory, so two agents editing files do not collide;
  * **hooks** that report progress upward, which is what feeds Agent Town.
Every failure path returns a reason. A delegation that silently does nothing is
worse than one that says "no key configured", because the user reads silence as
the agent deciding it had nothing to say.
"""

from __future__ import annotations

import os
import time
from dataclasses import dataclass, field
from typing import Any, Optional

#: How long one turn may take before we give up on it. Long, because an agent
#: that is planning and running code is legitimately slow, and short enough that
#: a wedged turn cannot hold the caller for a whole request timeout.
DEFAULT_TIMEOUT = 240.0

#: Where each agent's files live. Under the normal data dir so a Space restart
#: does not wipe an agent's work, which it would if this were under /tmp.
def _workspace_root() -> str:
    try:
        from core import data_paths
        base = data_paths.ensure_dir("agent_workspaces")
    except Exception:
        base = os.path.join(os.path.expanduser("~"), ".jarvis", "agent_workspaces")
        os.makedirs(base, exist_ok=True)
    return str(base)


def _looks_like_repr(text: str) -> bool:
    """True for a Python repr masquerading as an answer.
    '<Foo object at 0x7f...>' or 'Foo(...)'. The subagent output goes straight
    to a language model, which will happily write a confident paragraph about
    whatever it is handed — that is how a failed delegation became a confident
    "I audited the codebase and found nothing".
    """
    t = (text or "").strip()
    if not t:
        return False
    return (" object at 0x" in t or t.startswith("<") and t.endswith(">")
            or t.startswith("<google.") or t.startswith("<antigravity."))


# ── roles to capabilities ───────────────────────────────────────────────────
# The whole point of a roster is that a Copywriter and a Coder can be trusted
# with different things. Mapping that to a tool allowlist means the restriction
# is enforced by the runtime, not by a line in a prompt asking the model to
# behave. `*` is deliberately absent: an unknown role gets the smallest useful
# set, never everything.
ROLE_TOOLS: dict[str, tuple[str, ...]] = {
    "researcher": ("SEARCH_WEB", "READ_URL_CONTENT", "FIND_FILE", "READ_URL_CONTENT"),
    "scout":      ("SEARCH_WEB", "READ_URL_CONTENT", "SEARCH_DIR", "LIST_DIR"),
    "writer":     ("CREATE_FILE", "EDIT_FILE", "VIEW_FILE", "READ_URL_CONTENT"),
    "copywriter": ("CREATE_FILE", "EDIT_FILE", "VIEW_FILE"),
    "coder":      ("CREATE_FILE", "EDIT_FILE", "VIEW_FILE", "SEARCH_DIR",
                   "LIST_DIR", "FIND_FILE", "RUN_COMMAND", "FINISH"),
    "builder":    ("CREATE_FILE", "EDIT_FILE", "RUN_COMMAND", "VIEW_FILE",
                   "LIST_DIR", "FINISH"),
    "lead":       ("SEARCH_WEB", "READ_URL_CONTENT", "CREATE_FILE", "EDIT_FILE",
                   "VIEW_FILE", "SEARCH_DIR", "LIST_DIR", "FIND_FILE",
                   "RUN_COMMAND", "START_SUBAGENT", "FINISH"),
    "reporter":   ("SEARCH_WEB", "READ_URL_CONTENT", "CREATE_FILE"),
}
DEFAULT_ROLE_TOOLS = ("SEARCH_WEB", "READ_URL_CONTENT", "VIEW_FILE", "FINISH")

#: Roles allowed to run shell commands at all. Separate from the tool map so a
#: new role cannot accidentally inherit RUN_COMMAND by being added to the map.
SHELL_ROLES = frozenset({"coder", "builder", "lead"})


def tools_for(role: str, *, allow_shell: Optional[bool] = None) -> tuple[str, ...]:
    """The tool allowlist for a role.
    `allow_shell` is the override the approval layer uses: a user who approves a
    one-off command can grant it for that turn without editing the role table.
    """
    r = (role or "").strip().lower()
    tools = set(ROLE_TOOLS.get(r, DEFAULT_ROLE_TOOLS))
    if allow_shell is None:
        if r not in SHELL_ROLES:
            tools.discard("RUN_COMMAND")
    elif not allow_shell:
        tools.discard("RUN_COMMAND")
    return tuple(sorted(tools))


def _env(*names: str) -> str:
    for n in names:
        v = (os.environ.get(n) or "").strip()
        if v:
            return v
    return ""


def api_key() -> str:
    """The key, read the same way core/gemini.py reads it, so there is one
    answer to "is Gemini configured" rather than two that can disagree."""
    return _env("JARVIS_GEMINI_API_KEY", "GEMINI_API_KEY", "GOOGLE_API_KEY")


try:
    from google.antigravity.types import ModelEndpoint as _ME
    from google.antigravity.models import ModelTarget as _MT
except Exception:                                    # pragma: no cover
    _ME = _MT = None


class _BearerEndpoint(_ME):
    """A ModelEndpoint that carries an OpenAI-style bearer token.
    ModelEndpoint is abstract for one reason: validate_endpoint. Everything else
    it is, is a config holder for base_url and http_headers — and on the OpenAI
    connection those two fields are the whole authentication story, because
    LocalOpenAIAgentConfig has no api_key field to put a token in. Eleven lines
    to carry a token, after an afternoon of guessing which field the SDK wanted.
    """
    def validate_endpoint(self) -> None:
        if not (self.base_url or "").strip():
            raise ValueError("a gateway endpoint needs a base_url")


def _bearer_endpoint(base_url: str, token: str) -> Any:
    ep = _BearerEndpoint(base_url=base_url.rstrip("/"),
                         http_headers={"Authorization": f"Bearer {token}",
                                       "Content-Type": "application/json"})
    ep.validate_endpoint()
    return ep


def _model_target(name: str, endpoint: Any) -> Any:
    return _MT(name=name, endpoint=endpoint)


def _gateway_target() -> tuple[dict, str]:
    """({base_url, model, api_key}, description) for the OpenAI-compatible
    gateway, or (None, "").
    Gateway first, Gemini as the fallback — your instruction, and it is also the
    more robust arrangement. It cost us a day: the Gemini free tier is 20
    requests a day for the flash model, so a delegation that leans on Gemini
    primary stops working completely the moment that quota is gone, with nothing
    to fall back to. A gateway is a second provider with its own limits, so
    preferring it means one exhausted quota degrades throughput instead of
    ending delegation.
    """
    try:
        from core import gateway as G
        if not G.enabled():
            return None, ""
        url = (G.settings().get("openai_base_url") or "").strip().rstrip("/")
        name = G.chat_model() or "gpt-4o-mini"
        tok = (G.settings().get("openai_api_key")
               or G.settings().get("api_key") or "").strip()
        if not url:
            return None, ""
        return {"base_url": url, "model": name,
                "api_key": tok}, f"{name} via {url}"
    except Exception as e:
        return None, f"gateway present but unusable ({type(e).__name__}: {e})"


def sdk() -> Any:
    """The SDK module, or None. Lazy, because importing it pulls a runtime
    binary and nothing here should cost anything until it is actually used."""
    try:
        import google.antigravity as A
        return A
    except Exception:
        return None


def available() -> tuple[bool, str]:
    """(can we delegate?, why not). Never raises — callers show this to a user."""
    if not api_key():
        return False, ("no Gemini key. Set JARVIS_GEMINI_API_KEY or "
                       "GEMINI_API_KEY on the Space and I can delegate.")
    if sdk() is None:
        return False, ("the Antigravity SDK is not installed here. It is an "
                       "optional extra, so the app runs fine without it — but I "
                       "cannot delegate until it is present.")
    return True, "ready"


def capabilities() -> dict:
    """What this install can actually do. Reported by /api/health and rendered in
    the self-test, because "delegation is ready" and "delegation exists but has
    no key" are very different things to a person waiting on a result."""
    A = sdk()
    out: dict[str, Any] = {
        "sdk": bool(A), "key": bool(api_key()),
        "roles": sorted(ROLE_TOOLS), "sessions": False,
        "subagents": False, "budgets": False, "tools": [],
    }
    if not A:
        return out
    try:
        f = set(A.LocalAgentConfig.model_fields)
        out["sessions"] = "conversation_id" in f and "session_continuation_mode" in f
        out["subagents"] = "subagents" in f
        out["budgets"] = "budget_config" in f
        out["scopes"] = sorted(x for x in ("tools", "policies", "hooks",
                                          "workspaces", "skills_paths",
                                          "mcp_servers", "compaction_config",
                                          "retry_config") if x in f)
    except Exception:
        pass
    try:
        out["tools"] = sorted(m for m in dir(A.BuiltinTools) if m.isupper())
    except Exception:
        pass
    return out


# ── the unit of delegated work ──────────────────────────────────────────────
@dataclass
class Brief:
    """One piece of work handed to one agent.
    Deliberately not a bare string. The previous wrapper took `task: str` and
    that is why it could not scope a role, cap a budget, keep a session or
    report progress — there was nowhere to put any of that.
    """
    task: str
    role: str = ""
    name: str = ""
    agent_id: str = ""
    #: Stable per agent, so turn 40 remembers turn 1.
    session_id: str = ""
    #: "CREATE_OR_RESUME" by default: resume the agent's session if it exists,
    #: start one if not. Never CREATE_ONLY for ordinary work, because that
    #: throws away everything the agent knew.
    continuation: str = "CREATE_OR_RESUME"
    max_tool_calls: int = 40
    max_model_calls: int = 40
    max_total_tokens: int = 400_000
    allow_shell: Optional[bool] = None
    system_instructions: str = ""
    #: Native SubagentConfig list, for long-horizon work. Empty for a short
    #: task, and that is the correct answer rather than a default.
    subagents: list = field(default_factory=list)
    #: Populated by run(); the id to persist for next time.
    result_session: str = field(default="", init=False)

    def workspace(self) -> str:
        """A directory of this agent's own, created on demand."""
        slug = "".join(c if c.isalnum() or c in "-_" else "-"
                       for c in (self.agent_id or self.name or self.role or "agent")).strip("-")
        p = os.path.join(_workspace_root(), slug or "agent")
        os.makedirs(p, exist_ok=True)
        return p


def _continuation(A: Any, mode: str):
    """Map our string onto the SDK enum, tolerating an SDK that renames it."""
    enum = getattr(A, "SessionContinuationMode", None)
    if enum is None:
        return None
    return getattr(enum, mode, None)


def build_config(brief: Brief) -> tuple[Any, list[str]]:
    """Build a LocalAgentConfig for a brief. Returns (config, warnings).
    Warnings rather than exceptions: a missing optional capability should
    downgrade the turn, not cancel it.
    """
    A = sdk()
    assert A is not None, "build_config called with no SDK"
    warnings: list[str] = []
    kw: dict[str, Any] = {}

    # The key goes in explicitly. The SDK reads GEMINI_API_KEY from the
    # environment, which is a different name from ours, so a Space configured
    # with JARVIS_GEMINI_API_KEY looked ready and could not run anything.
    key = api_key()
    if key:
        kw["api_key"] = key

    # Gateway first, Gemini second — and the choice is made by which config
    # CLASS we build, not by a base_url string.
    #
    # I first pointed LocalAgentConfig at the gateway via GeminiAPIEndpoint and
    # got a 404 on /v1beta/models/...:streamGenerateContent: that connection
    # speaks the Gemini protocol, and the path is baked into bin/localharness,
    # a compiled binary, so no Python subclass can change it. The SDK already
    # ships the right one — LocalOpenAIAgentConfig, which takes base_url and
    # model directly and still carries subagents, tools, budgets, policies,
    # hooks and sessions. So the route is a class choice.
    target, why = _gateway_target()
    config_cls = A.LocalAgentConfig
    if target is not None:
        config_cls = getattr(A, "LocalOpenAIAgentConfig", None) or A.LocalAgentConfig
        kw["model"] = target["model"]
        if target.get("api_key"):
            # Point the SDK at the local bearer shim, not at the gateway.
            #
            # LocalOpenAIConnectionStrategy._build_harness_config constructs
            #     localharness_pb2.GemmaEndpoint(base_url=self._base_url)
            # and that proto has no auth, key, token or header field — the path
            # is built for unauthenticated local servers like llama.cpp. So a
            # token handed to LocalOpenAIAgentConfig, dropped on a ModelEndpoint
            # subclass, or set in the environment is discarded, and the gateway
            # answers 401 AUTH_002 every time.
            #
            # base_url is the one thing it does honour. So we give it a local
            # URL that is genuinely unauthenticated, and the shim adds the bearer
            # on the way out. This is the whole reason gw_shim exists.
            from core import gw_shim
            url = gw_shim.shared_url()
            if url:
                kw["base_url"] = url
            else:
                # No shim means no way to authenticate. Fall back to Gemini
                # rather than route to a gateway that will 401 on every call.
                warnings.append("gateway needs a token but the bearer shim could "
                                "not start — using Gemini directly")
                config_cls = A.LocalAgentConfig
                kw.pop("base_url", None)
        else:
            kw["base_url"] = target["base_url"]
    elif why:
        warnings.append(why + " — falling back to Gemini directly")

    if brief.system_instructions:
        kw["system_instructions"] = brief.system_instructions

    tools = list(tools_for(brief.role, allow_shell=brief.allow_shell))
    kw["tools"] = tools

    # BudgetConfig, like SubagentConfig, is in google.antigravity.types and not
    # on the module root. Getting that wrong raised, my except turned it into a
    # warning, and the warning was easy to miss — so every "spend ceiling" this
    # module claimed to set was quietly not set at all. A limit that is claimed
    # in a docstring and absent in the config is the same failure as a limit that
    # is absent and claimed in the config file.
    try:
        _types = __import__("google.antigravity.types", fromlist=["x"])
        kw["budget_config"] = _types.BudgetConfig(
            max_tool_calls=brief.max_tool_calls,
            max_model_calls=brief.max_model_calls,
            max_total_tokens=brief.max_total_tokens,
        )
    except Exception as e:
        warnings.append(f"NO SPEND CEILING ON THIS TURN ({e}) — an uncapped "
                        f"delegation is worse than a refused one")

    try:
        kw["workspaces"] = [brief.workspace()]
    except Exception as e:
        warnings.append(f"no private workspace ({e})")

    if brief.subagents:
        kw["subagents"] = brief.subagents

    if brief.session_id:
        kw["conversation_id"] = brief.session_id
        cont = _continuation(A, brief.continuation)
        if cont is not None:
            kw["session_continuation_mode"] = cont

    # Anything the installed SDK does not support is dropped with a warning
    # rather than raising: an SDK upgrade must not take delegation offline.
    # Against the class we are actually building, not always LocalAgentConfig:
    # base_url is a field of LocalOpenAIAgentConfig and not of LocalAgentConfig,
    # so checking the wrong one silently dropped the gateway address and the
    # turn died with 'requires a non-empty base_url'.
    supported = set(config_cls.model_fields)
    if kw:
        for k in [k for k in kw if k not in supported]:
            warnings.append(f"{k} is not supported by this SDK build — skipped")
            kw.pop(k, None)
    if target is not None:
        warnings.append(f"routing through the gateway: {why}")
    return config_cls(**kw), warnings


async def run_brief_async(brief: Brief, *, timeout: float = DEFAULT_TIMEOUT,
                          on_event=None) -> dict:
    """Run one brief to completion. `on_event(kind, data)` is called as the turn
    progresses; that is what makes a 90-second delegation show a spinner instead
    of looking like a hang."""
    import asyncio
    A = sdk()
    cfg, warnings = build_config(brief)
    for w in warnings:
        if on_event:
            on_event("warning", {"message": w})

    started = time.time()
    try:
        async with A.Agent(cfg) as agent:
            if on_event:
                on_event("started", {"name": brief.name or brief.role or "agent"})
            response = await asyncio.wait_for(agent.chat(brief.task), timeout=timeout)
            # There used to be a `except: text = str(response)` here.
            #
            # text() is a coroutine and it threw — and the fallback handed the
            # model "<google.antigravity.types.ChatResponse object at 0x7f...>"
            # as if it were the subagent's report. The model, being a model,
            # read that as a result and told the user it had audited the whole
            # codebase and found nothing. A Python repr is not an answer, and
            # a delegation that cannot produce one has failed.
            text = ""
            try:
                text = str(await response.text())
            except Exception as e:
                return {"ok": False, "session_id": brief.session_id,
                        "warnings": warnings,
                        "seconds": round(time.time() - started, 2),
                        "reason": (f"the agent ran but its answer could not be "
                                   f"read: {type(e).__name__}: {e}. The run "
                                   f"finished in "
                                   f"{round(time.time() - started, 2)}s.")}
            # Belt and braces. Even if the read succeeds, a string that is a
            # Python repr means we handed the caller an object, not a report.
            if _looks_like_repr(text):
                return {"ok": False, "session_id": brief.session_id,
                        "warnings": warnings,
                        "seconds": round(time.time() - started, 2),
                        "reason": ("the agent returned an object reference "
                                   "instead of a report — refusing to pass "
                                   "that off as an answer")}
            conv = ""
            for attr in ("conversation_id",):
                try:
                    conv = str(getattr(agent, attr, "") or "")
                except Exception:
                    conv = ""
            return {
                "ok": True, "text": text, "session_id": conv,
                "warnings": warnings,
                "seconds": round(time.time() - started, 2),
                "tools": list(cfg.tools or []),
            }
    except asyncio.TimeoutError:
        return {"ok": False, "session_id": brief.session_id, "warnings": warnings,
                "seconds": round(time.time() - started, 2),
                "reason": f"The agent did not finish within {int(timeout)}s."}
    except Exception as e:
        return {"ok": False, "session_id": brief.session_id, "warnings": warnings,
                "seconds": round(time.time() - started, 2),
                "reason": f"The agent failed on this task: {type(e).__name__}: {e}"}


def run_brief(brief: Brief, *, timeout: float = DEFAULT_TIMEOUT,
              on_event=None) -> dict:
    """Sync wrapper, because tool dispatch is synchronous.
    Tools are dispatched from a worker thread precisely so they cannot block the
    event loop, which means there is no loop running here and asyncio.run is the
    correct way in. If that ever stops being true this fails loudly with a clear
    reason instead of deadlocking.
    """
    task = (brief.task or "").strip()
    if not task:
        return {"ok": False, "reason": "I was given no task to delegate."}

    ok, why = available()
    if not ok:
        return {"ok": False, "reason": why}

    try:
        import asyncio
        asyncio.get_running_loop()
        return {"ok": False,
                "reason": ("Delegation has to run off the event loop. This call "
                           "was made from inside one, which is a bug in the "
                           "caller rather than something I can work around.")}
    except RuntimeError:
        pass

    brief.task = task
    try:
        return asyncio.run(run_brief_async(brief, timeout=timeout, on_event=on_event))
    except Exception as e:
        return {"ok": False, "reason": f"The agent runtime stopped: {e}"}


# ── back-compat ─────────────────────────────────────────────────────────────
async def run_turn_async(task: str) -> str:
    """The original one-turn-in, one-answer-out call. Kept because other modules
    import it; `run_brief` is what new code should use."""
    out = await run_brief_async(Brief(task=task, role="lead"))
    if not out.get("ok"):
        raise RuntimeError(out.get("reason") or "delegation failed")
    return out.get("text", "")


def run_turn(task: str, *, timeout: float = DEFAULT_TIMEOUT) -> dict:
    """Back-compat wrapper. New code should build a `Brief` so it can scope the
    role, cap the spend and keep the session."""
    return run_brief(Brief(task=task, role="lead"), timeout=timeout)

# ── the two shapes of delegated work ────────────────────────────────────────
# You said it plainly: JARVIS does the small things itself, and delegates the
# long-horizon ones to subagents that work autonomously and report back. The
# previous version modelled this as a workforce of job titles — a "coder", a
# "copywriter" — which is a different product. Nobody hires a Copywriter to
# fetch a URL. The unit is a task and a horizon, not a department.
SHORT = "short"      # one scoped turn. Cheap, fast, no fan-out.
LONG = "long"        # the agent plans, splits the work and runs subagents of its own.

#: What a subagent is for when the horizon is long. Each is a real independent
#: agent with its own instructions and its own narrow tool set, so the split is
#: enforced rather than described in a prompt.
SUBAGENT_ROLES: dict[str, tuple[str, tuple[str, ...]]] = {
    "gather": ("Collect the facts. Search, read, and report what you found with "
               "sources. Do not write anything or run anything.",
               ("SEARCH_WEB", "READ_URL_CONTENT")),
    "build":  ("Do the implementation work in your own workspace. Read before you "
               "write, and finish by saying what you changed.",
               ("VIEW_FILE", "CREATE_FILE", "EDIT_FILE", "LIST_DIR",
                "SEARCH_DIR", "FIND_FILE")),
    "verify": ("Check someone else's work. Read it, run it if there is a way to "
               "run it, and report what is actually wrong. Do not fix it.",
               ("VIEW_FILE", "LIST_DIR", "SEARCH_DIR", "READ_URL_CONTENT")),
}


@dataclass
class Delegation:
    """A piece of work handed to a subagent, and how much rope it gets.
    `horizon` decides the strategy. SHORT gets one scoped turn. LONG gets an
    agent that is allowed to split the work and run its own subagents, which is
    what "long-horizon" has to mean if a single 240-second turn is not enough.
    """
    task: str
    horizon: str = SHORT
    name: str = ""
    allow_shell: Optional[bool] = None
    #: Cap on how far the agent may fan out. A delegation that can spawn
    #: unboundedly is a delegation that can spend unboundedly.
    max_subagents: int = 3
    max_model_calls: int = 40
    max_tool_calls: int = 60
    max_total_tokens: int = 400_000
    timeout: float = DEFAULT_TIMEOUT
    system_instructions: str = ""

    @property
    def is_long(self) -> bool:
        return str(self.horizon or "").strip().lower() == LONG

    def role_tools(self) -> tuple[str, ...]:
        """A short delegation keeps the narrow toolset; a long one needs to be
        able to finish what it starts, including running what it wrote."""
        if self.is_long:
            return tools_for("lead", allow_shell=self.allow_shell)
        return tools_for("reporter", allow_shell=self.allow_shell)

    def subagents(self) -> list:
        """Native SubagentConfig list, or [] for a short task.
        Built from the SDK's own SubagentConfig rather than by asking the model
        in a prompt to "consider spawning helpers", because a request in a
        prompt is a suggestion and this is a capability list.
        """
        if not self.is_long or not self.max_subagents:
            return []
        A = sdk()
        if A is None:
            return []
        # SubagentConfig is not on the module root — it lives in
        # google.antigravity.types. Asking the root for it raised, my except
        # treated that as "no subagents available", and every long delegation
        # silently came back with an empty fan-out. A degrade that cannot be
        # distinguished from a normal result is worse than an error.
        try:
            from google.antigravity.types import SubagentConfig as SC
        except Exception:
            try:
                SC = A.SubagentConfig
            except Exception:
                return []
        out = []
        for i, (nm, (brief, tools)) in enumerate(SUBAGENT_ROLES.items()):
            if i >= self.max_subagents:
                break
            try:
                out.append(SC(name=nm, description=brief[:200],
                              system_instructions=brief, tools=list(tools)))
            except Exception:
                # A field this SDK build does not have is a downgrade, not a
                # reason to refuse the whole delegation.
                try:
                    out.append(SC(name=nm, description=brief[:200]))
                except Exception:
                    continue
        return out


def to_brief(d: Delegation, *, session_id: str = "") -> Brief:
    """A Delegation becomes a Brief plus its fan-out. Kept as an explicit
    conversion rather than making Brief and Delegation one type, because a Brief
    is one agent's turn and a Delegation is a decision about how much work that
    turn is allowed to be."""
    return Brief(
        task=d.task, role="lead" if d.is_long else "reporter",
        name=d.name or ("subagent" if d.is_long else "delegate"),
        session_id=session_id, allow_shell=d.allow_shell,
        max_tool_calls=d.max_tool_calls, max_model_calls=d.max_model_calls,
        max_total_tokens=d.max_total_tokens,
        system_instructions=d.system_instructions,
        subagents=d.subagents(),
    )


def run_delegation(d: Delegation, *, session_id: str = "",
                   on_event=None) -> dict:
    """Hand a task to a subagent. Short: one scoped turn. Long: an agent that
    may split the work across its own subagents and report back."""
    if not (d.task or "").strip():
        return {"ok": False, "reason": "I was given no task to delegate."}
    out = run_brief(to_brief(d, session_id=session_id),
                    timeout=d.timeout, on_event=on_event)
    out["horizon"] = d.horizon
    out["subagents"] = [getattr(x, "name", "?") for x in d.subagents()]
    return out