wehub-resource-sync
413 lines
16 KiB
Python
413 lines
16 KiB
Python
"""
|
|
LLM-callable async-dispatch builtins.
|
|
|
|
Step 11 of the harness contract introduces a small set of tools
|
|
the LLM uses to manage long-running work via the inbox pattern
|
|
instead of polling. This module ships:
|
|
|
|
- :class:`SysCallAsyncTool` (11a.i) — dispatch a local Python tool
|
|
as a background workflow.
|
|
- :class:`SysReadInboxTool` (11a.ii) — pull-mode drain of completed
|
|
async-work payloads.
|
|
- :class:`SysCancelAsyncTool` (11a.iii) — cancel a dispatched
|
|
task by its handle id; thin alias over the always-registered
|
|
``sys_cancel_task``.
|
|
|
|
All tools in this module are gated on the agent's top-level
|
|
``async:`` flag (see :attr:`AgentSpec.async_enabled`). The flag
|
|
**defaults to ``True``** to match the legacy inner stack's
|
|
default (``omnigent/inner/datamodel.py::AgentDef.async_enabled``),
|
|
so agents that don't mention it still see the async surface and
|
|
the same YAML produces the same tool list under Omnigent mode and the
|
|
legacy path. Agents that explicitly want a minimal-tools surface
|
|
declare ``async: false`` to suppress all three.
|
|
|
|
See ``designs/SERVER_HARNESS_CONTRACT.md`` §Async work + inbox
|
|
for the protocol design and the rationale for the inbox-vs-poll
|
|
flip.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from omnigent.tools.base import Tool
|
|
|
|
|
|
class SysCancelTaskTool(Tool):
|
|
"""
|
|
Cancel a background task by ``task_id``.
|
|
|
|
Non-blocking: marks the task cancelled and returns immediately.
|
|
The child workflow observes the cancel on its next runner-managed
|
|
iteration and emits the ``async_work_complete`` signal so the
|
|
parent's drain wakes.
|
|
|
|
This class defines the schema and name for ``sys_cancel_task``.
|
|
:class:`SysCancelAsyncTool` extends it with the ``handle_id``
|
|
alias schema.
|
|
"""
|
|
|
|
@classmethod
|
|
def name(cls) -> str:
|
|
""":returns: ``"sys_cancel_task"``."""
|
|
return "sys_cancel_task"
|
|
|
|
@classmethod
|
|
def description(cls) -> str:
|
|
""":returns: Human-readable description."""
|
|
return (
|
|
"Cancel a running background task. Non-blocking — the task "
|
|
"will transition to cancelled status; you'll see a "
|
|
"[System: task ... cancelled] message before your next "
|
|
"iteration. Already-terminal tasks are unchanged (no error). "
|
|
"NOTE: this tool only cancels tasks that Omnigent created via "
|
|
"sys_session_send / sys_call_async. Background shell commands "
|
|
"launched by Bash (`&`) are tracked by the Claude SDK — kill "
|
|
"those with the KillBash tool instead. Calling sys_cancel_task "
|
|
"on a Bash background id returns task_not_found."
|
|
)
|
|
|
|
def get_schema(self) -> dict[str, Any]:
|
|
""":returns: OpenAI function-format tool schema."""
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": self.name(),
|
|
"description": self.description(),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"task_id": {
|
|
"type": "string",
|
|
"description": (
|
|
"The task identifier to cancel — must be "
|
|
"an AP-created background task (from "
|
|
"sys_session_send / sys_call_async). NOT a "
|
|
"Bash background task id (those go to "
|
|
"KillBash)."
|
|
),
|
|
},
|
|
},
|
|
"required": ["task_id"],
|
|
},
|
|
},
|
|
}
|
|
|
|
def invoke(self, arguments: str, ctx: Any) -> str:
|
|
"""
|
|
Cancel the requested task.
|
|
|
|
The tasks table has been removed. This tool returns a
|
|
``task_not_found`` response for all inputs since no tasks are
|
|
persisted server-side.
|
|
|
|
:param arguments: JSON-encoded ``{"task_id": "..."}`` string.
|
|
:param ctx: Server-side execution context (unused).
|
|
:returns: JSON ``{"error": "task_not_found", ...}``.
|
|
"""
|
|
import json as _json
|
|
|
|
try:
|
|
args = _json.loads(arguments) if arguments else {}
|
|
except (_json.JSONDecodeError, ValueError):
|
|
args = {}
|
|
task_id = args.get("task_id", "")
|
|
return _json.dumps(
|
|
{
|
|
"error": "task_not_found",
|
|
"task_id": task_id,
|
|
"hint": (
|
|
"The tasks table has been removed. sys_cancel_task is "
|
|
"only effective for tasks created via sys_call_async "
|
|
"or sys_session_send; no server-persisted tasks exist."
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
class SysCallAsyncTool(Tool):
|
|
"""
|
|
Dispatch any local Python tool as a background task.
|
|
|
|
The LLM passes the target tool's name and a JSON-encoded
|
|
arguments string; this meta-tool spawns the target via the
|
|
runner-side async-task machinery that powers the
|
|
legacy ``@tool(synchronous=False)`` decoration. The handle
|
|
returned to the LLM describes the TARGET tool's task — the
|
|
LLM never sees a handle for ``sys_call_async`` itself.
|
|
|
|
Limitations (intentional v1):
|
|
|
|
- Only **local Python tools** can be dispatched. MCP tools,
|
|
builtins, and client-side tools fall through with an
|
|
``unknown_tool`` or ``unsupported_tool`` error. The
|
|
restriction matches the inner Session's pre-existing async
|
|
surface and keeps the dispatch path narrow.
|
|
- ``sys_call_async`` itself is NOT a valid target. Recursive
|
|
meta-dispatch would let the LLM build infinite handle chains
|
|
with no useful semantic; reject explicitly.
|
|
|
|
The handle round-trip is identical to the existing async
|
|
dispatch path (see
|
|
:class:`~omnigent.runtime.workflow._AsyncToolHandle`):
|
|
|
|
- ``task_id`` — the freshly created child task's id.
|
|
- ``tool_name`` — the TARGET tool's name (not
|
|
``"sys_call_async"``).
|
|
- ``status`` — ``"in_progress"``.
|
|
- ``message`` — the canonical async-handle instruction
|
|
string the LLM keys off of (G12).
|
|
"""
|
|
|
|
@classmethod
|
|
def name(cls) -> str:
|
|
""":returns: ``"sys_call_async"``."""
|
|
return "sys_call_async"
|
|
|
|
@classmethod
|
|
def description(cls) -> str:
|
|
""":returns: Human-readable description for the LLM."""
|
|
return (
|
|
"Dispatch a local Python tool as a background task. "
|
|
"Returns a task handle immediately; the result auto-"
|
|
"delivers as a system message when ready (or call "
|
|
"sys_read_inbox to drain proactively when that lands "
|
|
"in 11a.ii). Use this when you want to run a normally-"
|
|
"synchronous tool concurrently with other work — e.g., "
|
|
"kicking off several long calls in parallel."
|
|
)
|
|
|
|
def get_schema(self) -> dict[str, Any]:
|
|
"""
|
|
Return the OpenAI-format tool schema.
|
|
|
|
:returns: Dict with ``"type": "function"`` and a
|
|
``"function"`` sub-dict.
|
|
"""
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": SysCallAsyncTool.name(),
|
|
"description": SysCallAsyncTool.description(),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"tool": {
|
|
"type": "string",
|
|
"description": (
|
|
"Name of the local Python tool to "
|
|
"dispatch. Must match a tool the "
|
|
"agent has registered (see your "
|
|
"available tools)."
|
|
),
|
|
},
|
|
"args": {
|
|
"type": "string",
|
|
"description": (
|
|
"JSON-encoded arguments for the "
|
|
"target tool, e.g. "
|
|
'\'{"q": "weather"}\'. Same shape '
|
|
"the target tool would accept if "
|
|
"called directly."
|
|
),
|
|
},
|
|
},
|
|
"required": ["tool", "args"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
}
|
|
|
|
def is_async(self, arguments: str | None = None) -> bool:
|
|
"""
|
|
Always returns ``True`` — the whole point of this tool is
|
|
async dispatch.
|
|
|
|
Telling the runtime the tool is async-dispatching makes
|
|
AP's ``_call_tool`` route to :meth:`dispatch_async` instead
|
|
of :meth:`invoke`. The LLM-facing function_call_output
|
|
carries the resulting :class:`_AsyncToolHandle` JSON.
|
|
|
|
:param arguments: Ignored — async-ness here is intrinsic
|
|
to the tool, not derived from arguments. Kept for
|
|
interface parity with :class:`Tool.is_async`.
|
|
:returns: Always ``True``.
|
|
"""
|
|
return True
|
|
|
|
|
|
class SysReadInboxTool(Tool):
|
|
"""
|
|
Drain the parent task's inbox of completed async-work payloads.
|
|
|
|
The LLM calls this proactively to pull every
|
|
``async_work_complete`` payload that has piled up on the
|
|
parent workflow's inbox since the last drain. The
|
|
returned text is a concatenation of ``[System: task ...]``
|
|
blocks — the same format AP's between-iteration auto-collect
|
|
persists as user messages — but delivered inline as a
|
|
``function_call_output`` so the LLM doesn't have to wait for
|
|
the next iteration boundary.
|
|
|
|
Why an explicit tool: AP's auto-collect drain runs at the top
|
|
of every loop iteration (see ``_drain_async_completions``).
|
|
That's enough on its own when the LLM finishes a turn and the
|
|
framework gets another shot. But mid-turn, a long chain of
|
|
function_calls could complete several async tasks before the
|
|
LLM yields, and the LLM may want to inspect those completions
|
|
in the SAME turn — e.g., to fan out a second wave of work
|
|
based on the first wave's results. ``sys_read_inbox`` is the
|
|
pull-mode counterpart to the auto-collect push.
|
|
|
|
Consumes payloads off the topic — the next iteration's
|
|
auto-collect won't re-deliver them, so the LLM never sees the
|
|
same completion twice. (Inner had the same semantics; this
|
|
matches.)
|
|
|
|
Gated on ``async: true`` at the spec level (see
|
|
:class:`SysCallAsyncTool` docstring).
|
|
"""
|
|
|
|
@classmethod
|
|
def name(cls) -> str:
|
|
""":returns: ``"sys_read_inbox"``."""
|
|
return "sys_read_inbox"
|
|
|
|
@classmethod
|
|
def description(cls) -> str:
|
|
""":returns: Human-readable description for the LLM."""
|
|
return (
|
|
"Drain the inbox of completed async-work payloads "
|
|
"(from sys_call_async dispatches and sub-agent runs) "
|
|
"and return them inline. Use this mid-turn when you "
|
|
"want to inspect completions before yielding — e.g., "
|
|
"to plan follow-up work based on the results. Returns "
|
|
"a textual summary; an empty inbox returns a sentinel "
|
|
"string."
|
|
)
|
|
|
|
def get_schema(self) -> dict[str, Any]:
|
|
"""
|
|
Return the OpenAI-format tool schema.
|
|
|
|
:returns: Dict with ``"type": "function"`` and a
|
|
``"function"`` sub-dict. The tool takes no arguments.
|
|
"""
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": SysReadInboxTool.name(),
|
|
"description": SysReadInboxTool.description(),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {},
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
}
|
|
|
|
def is_async(self, arguments: str | None = None) -> bool:
|
|
"""
|
|
Always returns ``True``.
|
|
|
|
The drain reads from the parent workflow's async-completion
|
|
topic — an async-only API. AP's sync ``_call_tool`` path runs
|
|
in ``run_in_executor`` (a thread without an event loop), so
|
|
we can't call the drain from there. Returning ``True`` here
|
|
routes through :meth:`dispatch_async`, which Omnigent awaits
|
|
directly in the workflow's async body.
|
|
|
|
Despite the ``True`` return, this tool does NOT spawn a
|
|
child workflow — :meth:`dispatch_async` returns the result
|
|
as a string instead of an :class:`_AsyncToolHandle`. AP's
|
|
``_execute_tools`` accepts both shapes (see the
|
|
``isinstance(dispatched, _AsyncToolHandle)`` branch).
|
|
|
|
:param arguments: Ignored — the drain has no per-call
|
|
knobs.
|
|
:returns: Always ``True``.
|
|
"""
|
|
return True
|
|
|
|
|
|
class SysCancelAsyncTool(SysCancelTaskTool):
|
|
"""
|
|
Cancel an async-dispatched task by its handle id.
|
|
|
|
LLM-facing alias for :class:`SysCancelTaskTool` scoped to the
|
|
async-handle namespace. The schema takes ``handle_id`` instead
|
|
of ``task_id`` so the LLM's mental model — "I have a handle
|
|
from ``sys_call_async``; cancel it via ``sys_cancel_async``" —
|
|
maps cleanly to the tool list. Behaviour is identical to the
|
|
parent class: the parent's per-kind cancel primitives
|
|
(terminal SIGINT, ``client_tool`` SSE cancel) are inherited
|
|
unchanged.
|
|
|
|
Why a subclass and not a re-registration of
|
|
:class:`SysCancelTaskTool` under a second name: the schema's
|
|
parameter name differs (``handle_id`` vs ``task_id``), so the
|
|
JSON shape sent by the LLM is keyed differently. A subclass
|
|
cleanly inherits the cancel logic while overriding only the
|
|
schema-and-arg-translation surface.
|
|
|
|
Gated on ``async: true`` (see :class:`SysCallAsyncTool`'s
|
|
docstring). The generic :class:`SysCancelTaskTool` is always
|
|
registered via ``_register_task_lifecycle_tools``; this alias
|
|
is the async-namespace counterpart.
|
|
"""
|
|
|
|
@classmethod
|
|
def name(cls) -> str:
|
|
""":returns: ``"sys_cancel_async"``."""
|
|
return "sys_cancel_async"
|
|
|
|
@classmethod
|
|
def description(cls) -> str:
|
|
""":returns: Human-readable description for the LLM."""
|
|
return (
|
|
"Cancel a task you previously dispatched via "
|
|
"sys_call_async, using the handle id (the value of the "
|
|
"task_id field from the handle JSON). Non-blocking — "
|
|
"the task transitions to cancelled status and a "
|
|
"[System: task ... cancelled] block arrives in the "
|
|
"inbox or auto-deliver. Already-terminal tasks return "
|
|
"without changing state."
|
|
)
|
|
|
|
def get_schema(self) -> dict[str, Any]:
|
|
"""
|
|
Return the OpenAI-format tool schema.
|
|
|
|
Differs from :meth:`SysCancelTaskTool.get_schema` only in
|
|
the parameter name (``handle_id`` vs ``task_id``) and
|
|
``additionalProperties: false`` to match the rest of the
|
|
async-namespace schemas.
|
|
|
|
:returns: Dict with ``"type": "function"`` and a
|
|
``"function"`` sub-dict.
|
|
"""
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": SysCancelAsyncTool.name(),
|
|
"description": SysCancelAsyncTool.description(),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"handle_id": {
|
|
"type": "string",
|
|
"description": (
|
|
"The handle's task_id — same value "
|
|
"as the ``task_id`` field of the "
|
|
"handle JSON returned by "
|
|
"sys_call_async."
|
|
),
|
|
},
|
|
},
|
|
"required": ["handle_id"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
}
|