Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/autoskillit/server/tools/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ MCP `@mcp.tool()` handlers registered on import (20 tool modules).
| `__init__.py` | Docstring-only β€” tools register via `@mcp.tool()` on import |
| `_auto_overrides.py` | Shared `_build_auto_overrides()` factory for server-authoritative ingredient injection |
| `_cancellation_shield.py` | `_cancellation_shield` decorator β€” catches `asyncio.CancelledError` at MCP tool boundary, returns structured JSON |
| `_types.py` | TypedDict definitions for server tool JSON responses (RunSkillResult, RunCmdResult, etc.) |
| `_types.py` | TypedDict definitions for server tool JSON responses (RunSkillResult, RunCmdResult, ToolFailureEnvelope, etc.) and failure envelope factory helpers |
| `tools_kitchen.py` | `open_kitchen`, `close_kitchen` (gate lifecycle), `recipe://` MCP resource |
| `tools_config.py` | `configure_fleet`, `configure_order` (session config overlay) |
| `tools_agents.py` | `unlock_agent_pack` tool + `agent://` resource templates |
Expand Down
55 changes: 54 additions & 1 deletion src/autoskillit/server/tools/_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

from __future__ import annotations

from typing import Any, TypedDict
from typing import Any, Literal, TypedDict

from autoskillit.core import ModelTotalEntry, RetryReason

Expand All @@ -20,6 +20,9 @@
"TimingSummaryResult",
"KitchenStatusResult",
"DispatchEnvelopeResult",
"ToolFailureEnvelope",
"server_failure_envelope",
"input_failure_envelope",
]


Expand Down Expand Up @@ -190,3 +193,53 @@ class DispatchEnvelopeResult(TypedDict, total=False):
error: str
user_visible_message: str
details: dict[str, Any] | None


class _ToolFailureEnvelopeRequired(TypedDict):
"""Required fields for the structured failure envelope."""

success: Literal[False]
error: str
stage: str
retriable: bool


class ToolFailureEnvelope(_ToolFailureEnvelopeRequired, total=False):
"""Typed failure envelope with retriable discriminator for orchestrator routing.

Distinct from ``_kitchen_failure_envelope`` in tools_kitchen.py which returns
a raw JSON string with a ``kitchen`` field. This TypedDict provides a typed
dict contract with a ``retriable`` discriminator for P5-A4 orchestrator routing.
"""

user_visible_message: str


def server_failure_envelope(
exc: BaseException,
stage: str,
) -> ToolFailureEnvelope:
"""Build a failure envelope for server/infrastructure errors (retriable=True)."""
return ToolFailureEnvelope(
success=False,
error=f"{type(exc).__name__}: {exc}",
stage=stage,
retriable=True,
user_visible_message=(
f"Server error during {stage}: {type(exc).__name__}. "
"This may be transient β€” the orchestrator may retry."
),
)


def input_failure_envelope(
message: str,
stage: str,
) -> ToolFailureEnvelope:
"""Build a failure envelope for input/validation errors (retriable=False)."""
return ToolFailureEnvelope(
success=False,
error=message,
stage=stage,
retriable=False,
)
2 changes: 1 addition & 1 deletion tests/server/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,7 @@ Server tool handler unit tests β€” kitchen, execution, CI, clone, workspace tool
| `test_tools_status_mcp_response.py` | Tests for MCP response tracking integration in tools_status handlers |
| `test_tools_status_quota_and_db.py` | Tests for server status tools: quota events, telemetry writing, and DB access |
| `test_tools_status_summaries.py` | Tests for server status tools: token and timing summaries |
| `test_tools_types_module.py` | Tests for server/tools/_types.py TypedDict imports |
| `test_tools_types_module.py` | Tests for server/tools/_types.py TypedDict imports and failure envelope factory functions |
| `test_tools_workspace.py` | Tests for autoskillit server workspace tools |
| `test_tools_agents.py` | Tests for agent pack registry, MCP resources, and `unlock_agent_pack` |
| `test_track_response_size.py` | Tests for the track_response_size decorator in autoskillit.server._notify |
Expand Down
73 changes: 73 additions & 0 deletions tests/server/test_tools_types_module.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,3 +65,76 @@ def test_model_total_entry_still_in_core(self):
from autoskillit.core.types._type_results import ModelTotalEntry

assert "model" in ModelTotalEntry.__required_keys__


class TestToolFailureEnvelope:
"""Verify ToolFailureEnvelope TypedDict and factory helpers."""

def test_tool_failure_envelope_importable(self):
from autoskillit.server.tools._types import ToolFailureEnvelope

assert "success" in ToolFailureEnvelope.__required_keys__

def test_tool_failure_envelope_required_keys(self):
from autoskillit.server.tools._types import ToolFailureEnvelope

expected = {"success", "error", "stage", "retriable"}
assert expected <= ToolFailureEnvelope.__required_keys__

def test_tool_failure_envelope_optional_keys(self):
from autoskillit.server.tools._types import ToolFailureEnvelope

assert "user_visible_message" in ToolFailureEnvelope.__optional_keys__

def test_tool_failure_envelope_success_type(self):
import typing

from autoskillit.server.tools._types import ToolFailureEnvelope

hints = typing.get_type_hints(ToolFailureEnvelope)
assert typing.get_args(hints["success"]) == (False,)

def test_server_failure_envelope_factory(self):
from autoskillit.server.tools._types import server_failure_envelope

result = server_failure_envelope(ValueError("boom"), "init")
assert result["success"] is False
assert result["retriable"] is True
assert result["stage"] == "init"
assert "ValueError" in result["error"]
assert "boom" in result["error"]

def test_input_failure_envelope_factory(self):
from autoskillit.server.tools._types import input_failure_envelope

result = input_failure_envelope("bad input", "validate")
assert result["success"] is False
assert result["retriable"] is False
assert result["stage"] == "validate"
assert result["error"] == "bad input"

def test_server_failure_envelope_user_visible_message(self):
from autoskillit.server.tools._types import server_failure_envelope

result = server_failure_envelope(RuntimeError("oops"), "startup")
assert "user_visible_message" in result
assert len(result["user_visible_message"]) > 0

def test_tool_failure_envelope_in_all(self):
from autoskillit.server.tools._types import __all__ as types_all

assert "ToolFailureEnvelope" in types_all
assert "server_failure_envelope" in types_all
assert "input_failure_envelope" in types_all

def test_tool_failure_envelope_distinct_from_kitchen_envelope(self):
from autoskillit.server.tools._types import ToolFailureEnvelope
from autoskillit.server.tools.tools_kitchen import (
_kitchen_failure_envelope,
)

assert "retriable" in ToolFailureEnvelope.__required_keys__
all_keys = ToolFailureEnvelope.__required_keys__ | ToolFailureEnvelope.__optional_keys__
assert "kitchen" not in all_keys
result = _kitchen_failure_envelope(ValueError("x"), "test")
assert isinstance(result, str)
Loading