feat(api): enhance error handling and add structured hints for agent operations

This commit is contained in:
Abhimanyu Saharan
2026-02-15 02:00:54 +05:30
parent ccdff4835d
commit ee1cf05d5d
6 changed files with 1708 additions and 72 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -72,6 +72,47 @@ _RUNTIME_TYPE_REFERENCES = (UUID,)
AGENT_BOARD_ROLE_TAGS = cast("list[str | Enum]", ["agent-lead", "agent-worker"]) AGENT_BOARD_ROLE_TAGS = cast("list[str | Enum]", ["agent-lead", "agent-worker"])
def _agent_group_memory_openapi_hints(
*,
intent: str,
when_to_use: list[str],
routing_examples: list[dict[str, object]],
required_actor: str = "any_agent",
when_not_to_use: list[str] | None = None,
routing_policy: list[str] | None = None,
negative_guidance: list[str] | None = None,
prerequisites: list[str] | None = None,
side_effects: list[str] | None = None,
) -> dict[str, object]:
return {
"x-llm-intent": intent,
"x-when-to-use": when_to_use,
"x-when-not-to-use": when_not_to_use
or [
"Use a more specific endpoint when targeting a single actor or broadcast scope.",
],
"x-required-actor": required_actor,
"x-prerequisites": prerequisites
or [
"Authenticated actor token",
"Accessible board context",
],
"x-side-effects": side_effects
or ["Persisted memory visibility changes may be observable across linked boards."],
"x-negative-guidance": negative_guidance
or [
"Do not use as a replacement for direct task-specific commentary.",
"Do not assume infinite retention when group storage policies apply.",
],
"x-routing-policy": routing_policy
or [
"Use when board context requires shared memory discovery or posting.",
"Prefer narrow board endpoints for one-off lead/agent coordination needs.",
],
"x-routing-policy-examples": routing_examples,
}
def _parse_since(value: str | None) -> datetime | None: def _parse_since(value: str | None) -> datetime | None:
if not value: if not value:
return None return None
@@ -408,6 +449,27 @@ async def create_board_group_memory(
"", "",
response_model=DefaultLimitOffsetPage[BoardGroupMemoryRead], response_model=DefaultLimitOffsetPage[BoardGroupMemoryRead],
tags=AGENT_BOARD_ROLE_TAGS, tags=AGENT_BOARD_ROLE_TAGS,
openapi_extra=_agent_group_memory_openapi_hints(
intent="agent_board_group_memory_discovery",
when_to_use=[
"Inspect shared group memory for cross-board context before making decisions.",
"Collect active chat snapshots for a linked group before coordination actions.",
],
routing_examples=[
{
"input": {
"intent": "recover recent team memory for task framing",
"required_privilege": "agent_lead_or_worker",
},
"decision": "agent_board_group_memory_discovery",
}
],
side_effects=["No persisted side effects."],
routing_policy=[
"Use as a shared-context discovery step before decisioning.",
"Use board-specific memory endpoints for direct board persistence updates.",
],
),
) )
async def list_board_group_memory_for_board( async def list_board_group_memory_for_board(
*, *,
@@ -435,7 +497,31 @@ async def list_board_group_memory_for_board(
return await paginate(session, queryset.statement) return await paginate(session, queryset.statement)
@board_router.get("/stream", tags=AGENT_BOARD_ROLE_TAGS) @board_router.get(
"/stream",
tags=AGENT_BOARD_ROLE_TAGS,
openapi_extra=_agent_group_memory_openapi_hints(
intent="agent_board_group_memory_stream",
when_to_use=[
"Track shared group memory updates in near-real-time for live coordination.",
"React to newly added group messages without polling.",
],
routing_examples=[
{
"input": {
"intent": "subscribe to group memory updates for routing",
"required_privilege": "agent_lead_or_worker",
},
"decision": "agent_board_group_memory_stream",
}
],
side_effects=["No persisted side effects, streaming updates are read-only."],
routing_policy=[
"Use when coordinated decisions need continuous group context.",
"Prefer bounded history reads when a snapshot is sufficient.",
],
),
)
async def stream_board_group_memory_for_board( async def stream_board_group_memory_for_board(
request: Request, request: Request,
*, *,
@@ -472,7 +558,32 @@ async def stream_board_group_memory_for_board(
return EventSourceResponse(event_generator(), ping=15) return EventSourceResponse(event_generator(), ping=15)
@board_router.post("", response_model=BoardGroupMemoryRead, tags=AGENT_BOARD_ROLE_TAGS) @board_router.post(
"",
response_model=BoardGroupMemoryRead,
tags=AGENT_BOARD_ROLE_TAGS,
openapi_extra=_agent_group_memory_openapi_hints(
intent="agent_board_group_memory_record",
when_to_use=[
"Persist shared group memory for a linked group from board context.",
"Broadcast updates/messages to group-linked agents when chat or mention intent is present.",
],
routing_examples=[
{
"input": {
"intent": "share coordination signal in group memory",
"required_privilege": "board_agent",
},
"decision": "agent_board_group_memory_record",
}
],
side_effects=["Persist new group-memory entries with optional agent notification dispatch."],
routing_policy=[
"Use for shared memory writes that should be visible across linked boards.",
"Prefer direct board memory endpoints for board-local persistence.",
],
),
)
async def create_board_group_memory_for_board( async def create_board_group_memory_for_board(
payload: BoardGroupMemoryCreate, payload: BoardGroupMemoryCreate,
board: Board = BOARD_WRITE_DEP, board: Board = BOARD_WRITE_DEP,

View File

@@ -7,7 +7,7 @@ from datetime import datetime
from typing import Any from typing import Any
from uuid import UUID from uuid import UUID
from pydantic import field_validator from pydantic import ConfigDict, Field, field_validator
from sqlmodel import SQLModel from sqlmodel import SQLModel
from app.schemas.common import NonEmptyStr from app.schemas.common import NonEmptyStr
@@ -42,13 +42,64 @@ def _normalize_identity_profile(
class AgentBase(SQLModel): class AgentBase(SQLModel):
"""Common fields shared by agent create/read/update payloads.""" """Common fields shared by agent create/read/update payloads."""
board_id: UUID | None = None model_config = ConfigDict(
name: NonEmptyStr json_schema_extra={
status: str = "provisioning" "x-llm-intent": "agent_profile",
heartbeat_config: dict[str, Any] | None = None "x-when-to-use": [
identity_profile: dict[str, Any] | None = None "Create or update canonical agent metadata",
identity_template: str | None = None "Inspect agent attributes for governance or delegation",
soul_template: str | None = None ],
"x-when-not-to-use": [
"Task lifecycle operations (use task endpoints)",
"User-facing conversation content (not modeled here)",
],
"x-required-actor": "lead_or_worker_agent",
"x-prerequisites": [
"board_id if required by your board policy",
"identity templates should be valid JSON or text with expected markers",
],
"x-response-shape": "AgentRead",
"x-side-effects": [
"Reads or writes core agent profile fields",
"May impact routing or assignment decisions when persisted",
],
},
)
board_id: UUID | None = Field(
default=None,
description="Board id that scopes this agent. Omit only when policy allows global agents.",
examples=["11111111-1111-1111-1111-111111111111"],
)
name: NonEmptyStr = Field(
description="Human-readable agent display name.",
examples=["Ops triage lead"],
)
status: str = Field(
default="provisioning",
description="Current lifecycle state used by coordinator logic.",
examples=["provisioning", "active", "paused", "retired"],
)
heartbeat_config: dict[str, Any] | None = Field(
default=None,
description="Runtime heartbeat behavior overrides for this agent.",
examples=[{"interval_seconds": 30, "missing_tolerance": 120}],
)
identity_profile: dict[str, Any] | None = Field(
default=None,
description="Optional profile hints used by routing and policy checks.",
examples=[{"role": "incident_lead", "skill": "triage"}],
)
identity_template: str | None = Field(
default=None,
description="Template that helps define initial intent and behavior.",
examples=["You are a senior incident response lead."],
)
soul_template: str | None = Field(
default=None,
description="Template representing deeper agent instructions.",
examples=["When critical blockers appear, escalate in plain language."],
)
@field_validator("identity_template", "soul_template", mode="before") @field_validator("identity_template", "soul_template", mode="before")
@classmethod @classmethod
@@ -78,14 +129,66 @@ class AgentCreate(AgentBase):
class AgentUpdate(SQLModel): class AgentUpdate(SQLModel):
"""Payload for patching an existing agent.""" """Payload for patching an existing agent."""
board_id: UUID | None = None model_config = ConfigDict(
is_gateway_main: bool | None = None json_schema_extra={
name: NonEmptyStr | None = None "x-llm-intent": "agent_profile_update",
status: str | None = None "x-when-to-use": [
heartbeat_config: dict[str, Any] | None = None "Patch mutable agent metadata without replacing the full payload",
identity_profile: dict[str, Any] | None = None "Update status, templates, or heartbeat policy",
identity_template: str | None = None ],
soul_template: str | None = None "x-when-not-to-use": [
"Creating an agent (use AgentCreate)",
"Hard deletes or archive actions (use lifecycle endpoints)",
],
"x-required-actor": "board_lead",
"x-prerequisites": [
"Target agent id must exist and be visible to actor context",
],
"x-side-effects": [
"Mutates agent profile state",
],
},
)
board_id: UUID | None = Field(
default=None,
description="Optional new board assignment.",
examples=["22222222-2222-2222-2222-222222222222"],
)
is_gateway_main: bool | None = Field(
default=None,
description="Whether this agent is treated as the board gateway main.",
)
name: NonEmptyStr | None = Field(
default=None,
description="Optional replacement display name.",
examples=["Ops triage lead"],
)
status: str | None = Field(
default=None,
description="Optional replacement lifecycle status.",
examples=["active", "paused"],
)
heartbeat_config: dict[str, Any] | None = Field(
default=None,
description="Optional heartbeat policy override.",
examples=[{"interval_seconds": 45}],
)
identity_profile: dict[str, Any] | None = Field(
default=None,
description="Optional identity profile update values.",
examples=[{"role": "coordinator"}],
)
identity_template: str | None = Field(
default=None,
description="Optional replacement identity template.",
examples=["Focus on root cause analysis first."],
)
soul_template: str | None = Field(
default=None,
description="Optional replacement soul template.",
examples=["Escalate only after checking all known mitigations."],
)
@field_validator("identity_template", "soul_template", mode="before") @field_validator("identity_template", "soul_template", mode="before")
@classmethod @classmethod
@@ -111,30 +214,102 @@ class AgentUpdate(SQLModel):
class AgentRead(AgentBase): class AgentRead(AgentBase):
"""Public agent representation returned by the API.""" """Public agent representation returned by the API."""
id: UUID model_config = ConfigDict(
gateway_id: UUID json_schema_extra={
is_board_lead: bool = False "x-llm-intent": "agent_profile_lookup",
is_gateway_main: bool = False "x-when-to-use": [
openclaw_session_id: str | None = None "Inspect live agent state for routing and ownership decisions",
last_seen_at: datetime | None ],
created_at: datetime "x-required-actor": "board_lead_or_worker",
updated_at: datetime "x-interpretation": "This is a read model; changes here should use update/lifecycle endpoints.",
},
)
id: UUID = Field(description="Agent UUID.")
gateway_id: UUID = Field(description="Gateway UUID that manages this agent.")
is_board_lead: bool = Field(
default=False,
description="Whether this agent is the board lead.",
)
is_gateway_main: bool = Field(
default=False,
description="Whether this agent is the primary gateway agent.",
)
openclaw_session_id: str | None = Field(
default=None,
description="Optional openclaw session token.",
examples=["sess_01J..."],
)
last_seen_at: datetime | None = Field(
default=None,
description="Last heartbeat timestamp.",
)
created_at: datetime = Field(description="Creation timestamp.")
updated_at: datetime = Field(description="Last update timestamp.")
class AgentHeartbeat(SQLModel): class AgentHeartbeat(SQLModel):
"""Heartbeat status payload sent by agents.""" """Heartbeat status payload sent by agents."""
status: str | None = None model_config = ConfigDict(
json_schema_extra={
"x-llm-intent": "agent_health_signal",
"x-when-to-use": [
"Send periodic heartbeat to indicate liveness",
],
"x-required-actor": "any_agent",
"x-response-shape": "AgentRead",
},
)
status: str | None = Field(
default=None,
description="Agent health status string.",
examples=["healthy", "offline", "degraded"],
)
class AgentHeartbeatCreate(AgentHeartbeat): class AgentHeartbeatCreate(AgentHeartbeat):
"""Heartbeat payload used to create an agent lazily.""" """Heartbeat payload used to create an agent lazily."""
name: NonEmptyStr model_config = ConfigDict(
board_id: UUID | None = None json_schema_extra={
"x-llm-intent": "agent_bootstrap",
"x-when-to-use": [
"First heartbeat from a non-provisioned worker should bootstrap identity.",
],
"x-required-actor": "agent",
"x-prerequisites": ["Agent auth token already validated"],
"x-response-shape": "AgentRead",
},
)
name: NonEmptyStr = Field(
description="Display name assigned during first heartbeat bootstrap.",
examples=["Ops triage lead"],
)
board_id: UUID | None = Field(
default=None,
description="Optional board context for bootstrap.",
examples=["33333333-3333-3333-3333-333333333333"],
)
class AgentNudge(SQLModel): class AgentNudge(SQLModel):
"""Nudge message payload for pinging an agent.""" """Nudge message payload for pinging an agent."""
message: NonEmptyStr model_config = ConfigDict(
json_schema_extra={
"x-llm-intent": "agent_nudge",
"x-when-to-use": [
"Prompt a specific agent to revisit or reprioritize work.",
],
"x-required-actor": "board_lead",
"x-response-shape": "AgentRead",
},
)
message: NonEmptyStr = Field(
description="Short message to direct an agent toward immediate attention.",
examples=["Please update the incident triage status for task T-001."],
)

View File

@@ -2,7 +2,49 @@
from __future__ import annotations from __future__ import annotations
from sqlmodel import Field, SQLModel from pydantic import ConfigDict, Field
from sqlmodel import SQLModel
class LLMErrorResponse(SQLModel):
"""Standardized LLM-facing error payload used by API contracts."""
model_config = ConfigDict(
json_schema_extra={
"title": "LLMErrorResponse",
"x-llm-intent": "llm_error_handling",
"x-when-to-use": [
"Structured, tool-facing API errors for agent workflows",
"Gateway handoff and delegated-task operations",
],
"x-required-actor": "agent",
"x-side-effects": [
"Returns explicit machine-readable error context",
"Includes request_id for end-to-end traceability",
],
},
)
detail: str | dict[str, object] | list[object] = Field(
description=(
"Error payload. Agents should rely on `code` when present and default "
"to `message` for fallback display."
),
examples=["Invalid payload for lead escalation.", {"code": "not_found", "message": "Agent not found."}],
)
request_id: str | None = Field(
default=None,
description="Request correlation identifier injected by middleware.",
)
code: str | None = Field(
default=None,
description="Optional machine-readable error code.",
examples=["gateway_unavailable", "dependency_validation_failed"],
)
retryable: bool | None = Field(
default=None,
description="Whether a client should retry the call after remediating transient conditions.",
)
class BlockedTaskDetail(SQLModel): class BlockedTaskDetail(SQLModel):

View File

@@ -5,7 +5,8 @@ from __future__ import annotations
from typing import Literal from typing import Literal
from uuid import UUID from uuid import UUID
from sqlmodel import Field, SQLModel from pydantic import ConfigDict, Field
from sqlmodel import SQLModel
from app.schemas.common import NonEmptyStr from app.schemas.common import NonEmptyStr
@@ -23,72 +24,255 @@ def _user_reply_tags() -> list[str]:
class GatewayLeadMessageRequest(SQLModel): class GatewayLeadMessageRequest(SQLModel):
"""Request payload for sending a message to a board lead agent.""" """Request payload for sending a message to a board lead agent."""
kind: Literal["question", "handoff"] = "question" model_config = ConfigDict(
correlation_id: str | None = None json_schema_extra={
content: NonEmptyStr "x-llm-intent": "lead_direct_message",
"x-when-to-use": [
"A board has an urgent tactical request that needs direct lead routing",
"You need a specific lead response before delegating work",
],
"x-when-not-to-use": [
"Broadcasting to many leads (use broadcast request)",
"Requesting end-user decisions (use ask-user request)",
],
"x-required-actor": "main_agent",
"x-response-shape": "GatewayLeadMessageResponse",
},
)
kind: Literal["question", "handoff"] = Field(
default="question",
description="Routing mode for lead messages.",
examples=["question", "handoff"],
)
correlation_id: str | None = Field(
default=None,
description="Optional correlation token shared across upstream and downstream systems.",
examples=["lead-msg-1234"],
)
content: NonEmptyStr = Field(
description="Human-readable body sent to lead agents.",
examples=["Please triage the highest-priority blocker on board X."],
)
# How the lead should reply (defaults are interpreted by templates). # How the lead should reply (defaults are interpreted by templates).
reply_tags: list[str] = Field(default_factory=_lead_reply_tags) reply_tags: list[str] = Field(
reply_source: str | None = "lead_to_gateway_main" default_factory=_lead_reply_tags,
description="Tags required by reply templates when the lead responds.",
examples=[["gateway_main", "lead_reply"]],
)
reply_source: str | None = Field(
default="lead_to_gateway_main",
description="Reply destination key for the orchestrator.",
examples=["lead_to_gateway_main"],
)
class GatewayLeadMessageResponse(SQLModel): class GatewayLeadMessageResponse(SQLModel):
"""Response payload for a lead-message dispatch attempt.""" """Response payload for a lead-message dispatch attempt."""
ok: bool = True model_config = ConfigDict(
board_id: UUID json_schema_extra={
lead_agent_id: UUID | None = None "x-llm-intent": "lead_direct_message_result",
lead_agent_name: str | None = None "x-when-to-use": [
lead_created: bool = False "Confirm lead routing outcome for a direct message request.",
],
"x-when-not-to-use": [
"Broadcast outcomes (use GatewayLeadBroadcastResponse)",
],
"x-required-actor": "gateway_main",
"x-interpretation": "Use to confirm handoff path and recipient lead context.",
},
)
ok: bool = Field(default=True, description="Whether dispatch was accepted.")
board_id: UUID = Field(description="Board receiving the message.")
lead_agent_id: UUID | None = Field(
default=None,
description="Resolved lead agent id when present.",
)
lead_agent_name: str | None = Field(
default=None,
description="Resolved lead agent display name.",
)
lead_created: bool = Field(
default=False,
description="Whether a lead fallback actor was created during routing.",
)
class GatewayLeadBroadcastRequest(SQLModel): class GatewayLeadBroadcastRequest(SQLModel):
"""Request payload for broadcasting a message to multiple board leads.""" """Request payload for broadcasting a message to multiple board leads."""
kind: Literal["question", "handoff"] = "question" model_config = ConfigDict(
correlation_id: str | None = None json_schema_extra={
content: NonEmptyStr "x-llm-intent": "lead_broadcast_message",
board_ids: list[UUID] | None = None "x-when-to-use": [
reply_tags: list[str] = Field(default_factory=_lead_reply_tags) "Multiple board leads need the same message",
reply_source: str | None = "lead_to_gateway_main" "Coordinating cross-board operational alerts",
],
"x-when-not-to-use": [
"Single lead response required (use direct message)",
"Personalized board-level instruction from agent context",
],
"x-required-actor": "main_agent",
"x-response-shape": "GatewayLeadBroadcastResponse",
},
)
kind: Literal["question", "handoff"] = Field(
default="question",
description="Broadcast intent. `question` asks for responses; `handoff` requests transfer.",
examples=["question", "handoff"],
)
correlation_id: str | None = Field(
default=None,
description="Optional correlation token shared with downstream handlers.",
examples=["broadcast-2026-02-14"],
)
content: NonEmptyStr = Field(
description="Message content distributed to selected board leads.",
examples=["Board-wide incident: prioritize risk triage on task set 14."],
)
board_ids: list[UUID] | None = Field(
default=None,
description="Optional explicit list of board IDs; omit for lead-scoped defaults.",
examples=[[ "11111111-1111-1111-1111-111111111111" ]],
)
reply_tags: list[str] = Field(
default_factory=_lead_reply_tags,
description="Tags required by reply templates when each lead responds.",
examples=[["gateway_main", "lead_reply"]],
)
reply_source: str | None = Field(
default="lead_to_gateway_main",
description="Reply destination key for broadcast responses.",
examples=["lead_to_gateway_main"],
)
class GatewayLeadBroadcastBoardResult(SQLModel): class GatewayLeadBroadcastBoardResult(SQLModel):
"""Per-board result entry for a lead broadcast operation.""" """Per-board result entry for a lead broadcast operation."""
board_id: UUID model_config = ConfigDict(
lead_agent_id: UUID | None = None json_schema_extra={
lead_agent_name: str | None = None "x-llm-intent": "lead_broadcast_status",
ok: bool = False "x-when-to-use": [
error: str | None = None "Reading per-board outcomes for retries/follow-up workflows",
],
"x-when-not-to-use": ["Global summary checks should use parent broadcast response"],
"x-interpretation": "Use this result object as a transport status for one board.",
},
)
board_id: UUID = Field(description="Target board id for this result.")
lead_agent_id: UUID | None = Field(
default=None,
description="Resolved lead agent id for the target board.",
)
lead_agent_name: str | None = Field(
default=None,
description="Resolved lead agent display name.",
)
ok: bool = Field(default=False, description="Whether this board delivery succeeded.")
error: str | None = Field(
default=None,
description="Failure reason if this board failed.",
)
class GatewayLeadBroadcastResponse(SQLModel): class GatewayLeadBroadcastResponse(SQLModel):
"""Aggregate response for a lead broadcast operation.""" """Aggregate response for a lead broadcast operation."""
ok: bool = True model_config = ConfigDict(
sent: int = 0 json_schema_extra={
failed: int = 0 "x-llm-intent": "lead_broadcast_summary",
"x-when-to-use": [
"Inspect final counters after attempting a multi-board send.",
],
"x-when-not-to-use": [
"Single-board directed lead message (use GatewayLeadMessageResponse)",
],
"x-required-actor": "lead_agent_or_router",
"x-interpretation": "Use sent/failed counters before considering retry logic.",
"x-response-shape": "List of GatewayLeadBroadcastBoardResult",
},
)
ok: bool = Field(default=True, description="Whether broadcast execution succeeded.")
sent: int = Field(default=0, description="Number of boards successfully messaged.")
failed: int = Field(default=0, description="Number of boards that failed messaging.")
results: list[GatewayLeadBroadcastBoardResult] = Field(default_factory=list) results: list[GatewayLeadBroadcastBoardResult] = Field(default_factory=list)
class GatewayMainAskUserRequest(SQLModel): class GatewayMainAskUserRequest(SQLModel):
"""Request payload for asking the end user via a main gateway agent.""" """Request payload for asking the end user via a main gateway agent."""
correlation_id: str | None = None model_config = ConfigDict(
content: NonEmptyStr json_schema_extra={
preferred_channel: str | None = None "x-llm-intent": "human_escalation_request",
"x-when-to-use": [
"Blocking decision requires explicit user input",
"Task flow requires preference confirmation or permission",
],
"x-required-actor": "lead_agent",
"x-response-shape": "GatewayMainAskUserResponse",
},
)
correlation_id: str | None = Field(
default=None,
description="Optional correlation token for tracing request/response flow.",
examples=["ask-user-001"],
)
content: NonEmptyStr = Field(
description="Prompt that should be asked to the human.",
examples=["Can we proceed with the proposed vendor budget increase?"],
)
preferred_channel: str | None = Field(
default=None,
description="Optional preferred messaging channel.",
examples=["chat", "email"],
)
# How the main agent should reply back into Mission Control # How the main agent should reply back into Mission Control
# (defaults interpreted by templates). # (defaults interpreted by templates).
reply_tags: list[str] = Field(default_factory=_user_reply_tags) reply_tags: list[str] = Field(
reply_source: str | None = "user_via_gateway_main" default_factory=_user_reply_tags,
description="Tags required for routing the user response.",
examples=[["gateway_main", "user_reply"]],
)
reply_source: str | None = Field(
default="user_via_gateway_main",
description="Reply destination key for user confirmation loops.",
examples=["user_via_gateway_main"],
)
class GatewayMainAskUserResponse(SQLModel): class GatewayMainAskUserResponse(SQLModel):
"""Response payload for user-question dispatch via gateway main agent.""" """Response payload for user-question dispatch via gateway main agent."""
ok: bool = True model_config = ConfigDict(
board_id: UUID json_schema_extra={
main_agent_id: UUID | None = None "x-llm-intent": "human_escalation_result",
main_agent_name: str | None = None "x-when-to-use": [
"Track completion and main-agent handoff after human escalation request.",
],
"x-when-not-to-use": [
"Regular lead routing outcomes (use lead message/broadcast responses)",
],
"x-required-actor": "lead_agent",
"x-interpretation": "Track whether ask was accepted and which main agent handled it.",
},
)
ok: bool = Field(default=True, description="Whether ask-user dispatch was accepted.")
board_id: UUID = Field(description="Board context used for the request.")
main_agent_id: UUID | None = Field(
default=None,
description="Resolved main agent id handling the ask.",
)
main_agent_name: str | None = Field(
default=None,
description="Resolved main agent display name.",
)

View File

@@ -16,6 +16,10 @@ def _op_description(schema: dict[str, object], *, path: str, method: str) -> str
return str(op.get("description", "")).strip() return str(op.get("description", "")).strip()
def _schema_by_name(schema: dict[str, object], name: str) -> dict[str, object]:
return schema["components"]["schemas"][name] # type: ignore[return-value]
def test_openapi_agent_role_tags_are_exposed() -> None: def test_openapi_agent_role_tags_are_exposed() -> None:
"""Role tags should be queryable without path-based heuristics.""" """Role tags should be queryable without path-based heuristics."""
schema = app.openapi() schema = app.openapi()
@@ -30,6 +34,21 @@ def test_openapi_agent_role_tags_are_exposed() -> None:
path="/api/v1/agent/boards/{board_id}/tasks", path="/api/v1/agent/boards/{board_id}/tasks",
method="get", method="get",
) )
assert "agent-main" in _op_tags(
schema,
path="/api/v1/agent/boards",
method="get",
)
assert "agent-main" in _op_tags(
schema,
path="/api/v1/agent/boards/{board_id}",
method="get",
)
assert "agent-main" in _op_tags(
schema,
path="/api/v1/agent/agents",
method="get",
)
assert "agent-main" in _op_tags( assert "agent-main" in _op_tags(
schema, schema,
path="/api/v1/agent/gateway/leads/broadcast", path="/api/v1/agent/gateway/leads/broadcast",
@@ -78,3 +97,121 @@ def test_openapi_agent_role_endpoint_descriptions_exist() -> None:
path="/api/v1/boards/{board_id}/group-snapshot", path="/api/v1/boards/{board_id}/group-snapshot",
method="get", method="get",
) )
def test_openapi_agent_tool_endpoints_include_llm_hints() -> None:
"""Tool-facing agent endpoints should expose structured usage hints and operation IDs."""
schema = app.openapi()
op_ids: set[str] = set()
expected_paths = [
("/api/v1/agent/boards", "get"),
("/api/v1/agent/boards/{board_id}", "get"),
("/api/v1/agent/agents", "get"),
("/api/v1/agent/heartbeat", "post"),
("/api/v1/agent/boards/{board_id}/tasks", "post"),
("/api/v1/agent/boards/{board_id}/tasks", "get"),
("/api/v1/agent/boards/{board_id}/tags", "get"),
("/api/v1/agent/boards/{board_id}/tasks/{task_id}", "patch"),
("/api/v1/agent/boards/{board_id}/tasks/{task_id}/comments", "get"),
("/api/v1/agent/boards/{board_id}/tasks/{task_id}/comments", "post"),
("/api/v1/agent/boards/{board_id}/memory", "get"),
("/api/v1/agent/boards/{board_id}/memory", "post"),
("/api/v1/boards/{board_id}/group-memory", "get"),
("/api/v1/boards/{board_id}/group-memory", "post"),
("/api/v1/boards/{board_id}/group-memory/stream", "get"),
("/api/v1/agent/boards/{board_id}/approvals", "get"),
("/api/v1/agent/boards/{board_id}/approvals", "post"),
("/api/v1/agent/boards/{board_id}/onboarding", "post"),
("/api/v1/agent/boards/{board_id}/agents/{agent_id}/soul", "get"),
("/api/v1/agent/agents", "post"),
("/api/v1/agent/boards/{board_id}/agents/{agent_id}/nudge", "post"),
("/api/v1/agent/boards/{board_id}/agents/{agent_id}/soul", "put"),
("/api/v1/agent/boards/{board_id}/agents/{agent_id}", "delete"),
("/api/v1/agent/boards/{board_id}/gateway/main/ask-user", "post"),
("/api/v1/agent/gateway/boards/{board_id}/lead/message", "post"),
("/api/v1/agent/gateway/leads/broadcast", "post"),
]
for path, method in expected_paths:
op = schema["paths"][path][method]
assert "x-llm-intent" in op
assert isinstance(op["x-llm-intent"], str)
assert op["x-llm-intent"]
assert "x-negative-guidance" in op
assert isinstance(op["x-negative-guidance"], list)
assert op["x-negative-guidance"]
assert all(isinstance(item, str) and item for item in op["x-negative-guidance"])
assert "x-when-to-use" in op
assert op["x-when-to-use"]
assert "x-routing-policy" in op
assert op["x-routing-policy"]
assert isinstance(op["x-routing-policy"], list)
assert op["x-routing-policy"]
assert all(isinstance(item, str) and item for item in op["x-routing-policy"])
assert "x-required-actor" in op
assert "operationId" in op
assert isinstance(op["operationId"], str)
assert op["operationId"]
assert "x-routing-policy-examples" in op
assert isinstance(op["x-routing-policy-examples"], list)
assert op["x-routing-policy-examples"]
assert all(
isinstance(example, dict)
and "decision" in example
and "input" in example
and isinstance(example["decision"], str)
and example["decision"].strip()
and isinstance(example["input"], dict)
and "intent" in example["input"]
and isinstance(example["input"]["intent"], str)
and example["input"]["intent"].strip()
for example in op["x-routing-policy-examples"]
)
op_ids.add(op["operationId"])
responses = op.get("responses", {})
assert responses
assert len(op_ids) == len(expected_paths)
def test_openapi_agent_schemas_include_discoverability_hints() -> None:
"""Schema-level metadata should advertise usage context for model-driven tooling."""
schema = app.openapi()
expected_schema_hints = [
("AgentCreate", "agent_profile"),
("AgentUpdate", "agent_profile_update"),
("AgentRead", "agent_profile_lookup"),
("GatewayLeadMessageRequest", "lead_direct_message"),
("GatewayLeadMessageResponse", "lead_direct_message_result"),
("GatewayLeadBroadcastResponse", "lead_broadcast_summary"),
("GatewayMainAskUserRequest", "human_escalation_request"),
("GatewayMainAskUserResponse", "human_escalation_result"),
("AgentNudge", "agent_nudge"),
]
for schema_name, intent in expected_schema_hints:
component = _schema_by_name(schema, schema_name)
assert "x-llm-intent" in component
assert component["x-llm-intent"] == intent
assert component.get("x-when-to-use")
assert component.get("x-required-actor") or component_name_is_query(schema_name)
def schema_name_is_query(schema_name: str) -> bool:
"""Some pure response shapes are actor-agnostic and expose interpretation instead."""
return schema_name in {"GatewayLeadBroadcastResponse", "GatewayMainAskUserResponse"}
def test_openapi_agent_schema_fields_have_context() -> None:
"""Request/response fields should include field-level usage hints."""
schema = app.openapi()
request_schema = _schema_by_name(schema, "GatewayLeadMessageRequest")
props = request_schema["properties"] # type: ignore[assignment]
assert "kind" in props
assert props["kind"]["description"]
assert props["kind"]["description"].startswith("Routing mode")
nudge_schema = _schema_by_name(schema, "AgentNudge")
nudge_props = nudge_schema["properties"] # type: ignore[assignment]
assert "message" in nudge_props
assert nudge_props["message"]["description"]