"""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. '' 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(" 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 "" # 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