feat: enforce runtime domain invariants (#45)

## What

- reject blank task, session, MCP, tool, artifact, and result identities
- require positive execution hints and finite non-negative
budgets/counts
- reject ambiguous session start/resume state, duplicate MCP names, and
conflicting tool filters
- canonicalize sequence value fields while rejecting scalar strings
- add AgentResult.is_success
- represent unknown Usage and cost as None while preserving reported
zero
- preserve partial provider telemetry without inventing cache counts
- fix Claude resume result fallback when the SDK omits its session id

## Why

Several invalid or ambiguous states were constructible, and missing
telemetry was indistinguishable from a real zero.

## Root cause

The frozen dataclasses provided immutability but did not own their
cross-field invariants, while adapter coercion helpers normalized
missing and malformed counters to zero.

## Checks

- 321 passed, 12 skipped
- independent domain audit completed
- ruff check src tests
- mypy
- uv lock --check
- wheel and sdist build

## Compatibility

This intentionally rejects previously accepted invalid constructor
inputs and changes unknown Usage fields from zero to None.

## Stack

- Base: #44
- Next: #46

Author: ebarti

CHANGELOG.md

@@ -15,11 +15,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   absent or rejected structured output.
 - All built-in adapters and public fake runtimes now validate returned values
   against `output_schema` locally, including textual JSON fallbacks.
+- `AgentResult.is_success` provides one canonical success predicate.
 
 ### Changed
 
 - Pydantic structured-output parsing now requests strict validation, so values
   such as `"42"` no longer coerce into integer fields.
+- BREAKING: task and value-object constructors now reject blank identities,
+  non-positive execution hints, invalid budgets, ambiguous session inputs,
+  duplicate MCP names, and conflicting tool filters.
+- BREAKING: unknown `Usage` values, including cost, are `None` rather than zero;
+  an explicit provider-reported zero remains `0`/`0.0`.
 
 ### Fixed
 

README.md

@@ -143,9 +143,11 @@ silently dropping it.
 
 `AgentResult` returns output, finish reason (see `FinishReason`), locally
 validated structured output, usage, cost, session id, tool-call audits, and
-provider metadata. `parsed_output_available` distinguishes an absent payload
-from valid JSON `null` (both use `parsed_output=None`). `artifacts` is a reserved
-field: no built-in runtime populates it yet, so it is always an empty tuple today.
+provider metadata. `is_success` is true only for a natural, error-free
+completion. Unknown usage and cost fields are `None`; reported zero remains
+distinct. `parsed_output_available` distinguishes an absent payload from valid
+JSON `null` (both use `parsed_output=None`). `artifacts` is a reserved field: no
+built-in runtime populates it yet, so it is always an empty tuple today.
 
 ## Docs
 

docs/api-stability.md

@@ -33,6 +33,13 @@ import the names from the top-level package instead.
   convention and compare by value. The wrappers remain plain `dict` subclasses,
   so `dataclasses.asdict`, `pickle`, `copy.deepcopy`, and JSON serialization
   keep working.
+- **Constructors enforce domain invariants.** Goals and identifiers are
+  non-blank; execution hints are positive; counts and monetary values are
+  finite and non-negative; session start and resume inputs are mutually
+  exclusive; MCP server names and tool filters are unambiguous.
+- **Unknown telemetry is not zero.** Every `Usage` field is nullable. `None`
+  means the provider did not report a value; `0` and `0.0` mean it explicitly
+  reported zero.
 - **Unsupported inputs raise, they are not dropped.** An adapter that cannot honor
   a task field raises `UnsupportedTaskInputError`; the one exception is
   vendor-option drift, which is recorded in

docs/providers.md

@@ -38,6 +38,10 @@ vocabulary (for example `claude-agent-sdk` 0.2.x accepts
 `low`/`medium`/`high`/`xhigh`/`max`), and an SDK too old to accept `effort` at
 all records the drop in `AgentResult.metadata["dropped_options"]`.
 
+Usage fields are `None` when a provider omits its usage breakdown or cost.
+Provider-reported zeros remain numeric zero, so callers can distinguish a free
+or zero-token run from missing telemetry.
+
 Claude uses the `claude-agent-sdk` package and maps working directory,
 permissions, filesystem access (a `READ_ONLY` filesystem forces `plan` mode),
 MCP servers, sessions, structured output, tool allow/deny lists, runtime

src/agent_runtime_kit/_types.py

@@ -5,6 +5,7 @@
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from enum import Enum
+from math import isfinite
 from pathlib import Path
 from typing import Any, Generic, NoReturn, Protocol, TypeVar, cast, runtime_checkable
 from uuid import uuid4
@@ -81,6 +82,44 @@ def _coerce_enum(enum_cls: type[_EnumT], value: Any, field_name: str) -> _EnumT:
         raise ValueError(f"invalid {field_name} {value!r}; valid values: {valid}") from None
 
 
+def _require_nonblank(value: str, field_name: str) -> None:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{field_name} must be a non-empty string")
+
+
+def _require_unique_nonblank(values: tuple[str, ...], field_name: str) -> None:
+    for value in values:
+        _require_nonblank(value, field_name)
+    duplicates = sorted({value for value in values if values.count(value) > 1})
+    if duplicates:
+        raise ValueError(f"{field_name} contains duplicates: {duplicates}")
+
+
+def _tuple_value(value: Any, field_name: str) -> tuple[Any, ...]:
+    if isinstance(value, (str, bytes)):
+        raise ValueError(f"{field_name} must be a sequence, not a scalar string")
+    try:
+        return tuple(value)
+    except TypeError as exc:
+        raise ValueError(f"{field_name} must be an iterable") from exc
+
+
+def _validate_optional_count(value: int | None, field_name: str) -> None:
+    if value is None:
+        return
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        raise ValueError(f"{field_name} must be a non-negative integer or None")
+
+
+def _validate_optional_amount(value: float | None, field_name: str) -> None:
+    if value is None:
+        return
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"{field_name} must be a non-negative finite number or None")
+    if value < 0 or not isfinite(float(value)):
+        raise ValueError(f"{field_name} must be a non-negative finite number or None")
+
+
 class AgentRuntimeKind(str, Enum):
     """Supported runtime families."""
 
@@ -257,6 +296,17 @@ class McpServerConfig:
     env: Mapping[str, str] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
+        _require_nonblank(self.name, "McpServerConfig.name")
+        _require_nonblank(self.command, "McpServerConfig.command")
+        object.__setattr__(self, "args", _tuple_value(self.args, "McpServerConfig.args"))
+        if not all(isinstance(arg, str) for arg in self.args):
+            raise ValueError("McpServerConfig.args must contain only strings")
+        if not isinstance(self.env, Mapping):
+            raise ValueError("McpServerConfig.env must be a mapping")
+        for key, value in self.env.items():
+            _require_nonblank(key, "McpServerConfig.env key")
+            if not isinstance(value, str):
+                raise ValueError("McpServerConfig.env values must be strings")
         object.__setattr__(self, "env", _freeze_mapping(self.env))
 
 
@@ -285,6 +335,25 @@ def __post_init__(self) -> None:
                 "filesystem",
                 _coerce_enum(FilesystemAccess, self.filesystem, "filesystem"),
             )
+        object.__setattr__(
+            self,
+            "allowed_tools",
+            _tuple_value(self.allowed_tools, "PermissionProfile.allowed_tools"),
+        )
+        object.__setattr__(
+            self,
+            "disallowed_tools",
+            _tuple_value(self.disallowed_tools, "PermissionProfile.disallowed_tools"),
+        )
+        _require_unique_nonblank(self.allowed_tools, "PermissionProfile.allowed_tools")
+        _require_unique_nonblank(self.disallowed_tools, "PermissionProfile.disallowed_tools")
+        overlap = sorted(set(self.allowed_tools) & set(self.disallowed_tools))
+        if overlap:
+            raise ValueError(
+                "PermissionProfile cannot both allow and disallow tools: " + ", ".join(overlap)
+            )
+        if self.network is not None and not isinstance(self.network, bool):
+            raise ValueError("PermissionProfile.network must be bool or None")
 
 
 @dataclass(frozen=True)
@@ -298,6 +367,14 @@ class ToolCallAudit:
     duration_ms: int = 0
 
     def __post_init__(self) -> None:
+        _require_nonblank(self.tool_name, "ToolCallAudit.tool_name")
+        _require_nonblank(self.status, "ToolCallAudit.status")
+        if (
+            isinstance(self.duration_ms, bool)
+            or not isinstance(self.duration_ms, int)
+            or self.duration_ms < 0
+        ):
+            raise ValueError("ToolCallAudit.duration_ms must be a non-negative integer")
         object.__setattr__(self, "arguments", _freeze_mapping(self.arguments))
 
 
@@ -310,6 +387,8 @@ class ArtifactRef:
     metadata: Mapping[str, Any] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
+        _require_nonblank(self.uri, "ArtifactRef.uri")
+        _require_nonblank(self.kind, "ArtifactRef.kind")
         object.__setattr__(self, "metadata", _freeze_mapping(self.metadata))
 
 
@@ -325,25 +404,39 @@ class SessionResumeState:
     session_id: str
     transcript: tuple[Any, ...] = ()
 
+    def __post_init__(self) -> None:
+        _require_nonblank(self.session_id, "SessionResumeState.session_id")
+        object.__setattr__(
+            self,
+            "transcript",
+            _tuple_value(self.transcript, "SessionResumeState.transcript"),
+        )
+
 
 @dataclass(frozen=True)
 class Usage:
     """Token and cost metadata reported by a runtime.
 
     ``input_tokens`` counts prompt tokens excluding Anthropic-style cache reads and
     cache creation, which are reported separately in ``cache_read_tokens`` and
-    ``cache_creation_tokens``. ``total_tokens`` is the vendor-reported total when the
-    runtime provides one, and ``None`` when it does not (so "unknown" is
-    distinguishable from zero). ``cost_usd`` is ``0.0`` when the provider reports no
-    cost.
+    ``cache_creation_tokens``. Every field is ``None`` when unknown, so a reported
+    zero remains distinguishable from missing telemetry.
     """
 
-    input_tokens: int = 0
-    output_tokens: int = 0
-    cache_read_tokens: int = 0
-    cache_creation_tokens: int = 0
+    input_tokens: int | None = None
+    output_tokens: int | None = None
+    cache_read_tokens: int | None = None
+    cache_creation_tokens: int | None = None
     total_tokens: int | None = None
-    cost_usd: float = 0.0
+    cost_usd: float | None = None
+
+    def __post_init__(self) -> None:
+        _validate_optional_count(self.input_tokens, "Usage.input_tokens")
+        _validate_optional_count(self.output_tokens, "Usage.output_tokens")
+        _validate_optional_count(self.cache_read_tokens, "Usage.cache_read_tokens")
+        _validate_optional_count(self.cache_creation_tokens, "Usage.cache_creation_tokens")
+        _validate_optional_count(self.total_tokens, "Usage.total_tokens")
+        _validate_optional_amount(self.cost_usd, "Usage.cost_usd")
 
 
 @dataclass(frozen=True)
@@ -375,6 +468,30 @@ class AgentTask:
     metadata: Mapping[str, Any] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
+        _require_nonblank(self.goal, "AgentTask.goal")
+        _require_nonblank(self.task_id, "AgentTask.task_id")
+        if self.model is not None:
+            _require_nonblank(self.model, "AgentTask.model")
+        if self.reasoning_effort is not None:
+            _require_nonblank(self.reasoning_effort, "AgentTask.reasoning_effort")
+        if isinstance(self.sdk_executions, bool) or not isinstance(self.sdk_executions, int):
+            raise ValueError("AgentTask.sdk_executions must be a positive integer")
+        if self.sdk_executions < 1:
+            raise ValueError("AgentTask.sdk_executions must be a positive integer")
+        _validate_optional_amount(self.budget_usd, "AgentTask.budget_usd")
+        if self.session_id is not None:
+            _require_nonblank(self.session_id, "AgentTask.session_id")
+        if self.session_id is not None and self.resume_from is not None:
+            raise ValueError("AgentTask.session_id and resume_from are mutually exclusive")
+        object.__setattr__(
+            self,
+            "mcp_servers",
+            _tuple_value(self.mcp_servers, "AgentTask.mcp_servers"),
+        )
+        server_names = tuple(server.name for server in self.mcp_servers)
+        duplicates = sorted({name for name in server_names if server_names.count(name) > 1})
+        if duplicates:
+            raise ValueError(f"AgentTask.mcp_servers contains duplicate names: {duplicates}")
         object.__setattr__(self, "metadata", _freeze_mapping(self.metadata))
         if self.output_schema is not None:
             # Local import avoids an import cycle: _schema's typed errors import
@@ -405,16 +522,39 @@ class AgentResult:
     parsed_output_available: bool = False
 
     def __post_init__(self) -> None:
+        _require_nonblank(self.finish_reason, "AgentResult.finish_reason")
+        if self.error is not None:
+            _require_nonblank(self.error, "AgentResult.error")
+            if self.finish_reason == FinishReason.DONE:
+                raise ValueError("AgentResult.error requires a non-success finish_reason")
+        if isinstance(self.rounds, bool) or not isinstance(self.rounds, int) or self.rounds < 0:
+            raise ValueError("AgentResult.rounds must be a non-negative integer")
+        object.__setattr__(
+            self,
+            "tool_calls",
+            _tuple_value(self.tool_calls, "AgentResult.tool_calls"),
+        )
+        object.__setattr__(
+            self,
+            "artifacts",
+            _tuple_value(self.artifacts, "AgentResult.artifacts"),
+        )
         object.__setattr__(self, "metadata", _freeze_mapping(self.metadata))
         if self.parsed_output is not None and not self.parsed_output_available:
             object.__setattr__(self, "parsed_output_available", True)
 
     @property
-    def cost_usd(self) -> float:
+    def cost_usd(self) -> float | None:
         """Return the reported task cost in USD."""
 
         return self.usage.cost_usd
 
+    @property
+    def is_success(self) -> bool:
+        """Whether the runtime completed naturally without an error."""
+
+        return self.finish_reason == FinishReason.DONE and self.error is None
+
 
 _ParsedT = TypeVar("_ParsedT")
 

src/agent_runtime_kit/adapters/_common.py

@@ -6,6 +6,7 @@
 import os
 from collections.abc import Iterable, Mapping
 from importlib import metadata, util
+from math import isfinite
 from typing import Any
 
 from agent_runtime_kit._errors import UnsupportedTaskInputError
@@ -259,13 +260,27 @@ def field_value(value: Any, name: str, default: Any = None) -> Any:
     return getattr(value, name, default)
 
 
-def optional_int(value: Any) -> int:
-    """Coerce a vendor-reported value to ``int``, treating ``None``/junk as 0."""
+def optional_int(value: Any) -> int | None:
+    """Coerce a non-negative vendor count, preserving unknown as None."""
 
+    if value is None or isinstance(value, bool):
+        return None
+    if isinstance(value, float):
+        if not isfinite(value) or not value.is_integer():
+            return None
+    if isinstance(value, str) and not value.strip():
+        return None
     try:
-        return int(value or 0)
-    except (TypeError, ValueError):
-        return 0
+        parsed = int(value)
+    except (TypeError, ValueError, OverflowError):
+        return None
+    if not isinstance(value, (int, float, str)):
+        try:
+            if value != parsed:
+                return None
+        except Exception:
+            return None
+    return parsed if parsed >= 0 else None
 
 
 def optional_str(value: Any) -> str | None:

src/agent_runtime_kit/adapters/antigravity.py

@@ -918,14 +918,24 @@ def _default_data_dir() -> Path:
 
 
 def _usage_from(value: Any) -> Usage:
+    if value is None:
+        return Usage()
     prompt_tokens = optional_int(getattr(value, "prompt_token_count", None))
     output_tokens = optional_int(getattr(value, "candidates_token_count", None))
     thoughts = optional_int(getattr(value, "thoughts_token_count", None))
     cache_read = optional_int(getattr(value, "cached_content_token_count", None))
     total = optional_int(getattr(value, "total_token_count", None))
     return Usage(
-        input_tokens=max(prompt_tokens - cache_read, 0),
-        output_tokens=output_tokens + thoughts,
+        input_tokens=(
+            max(prompt_tokens - cache_read, 0)
+            if prompt_tokens is not None and cache_read is not None
+            else None
+        ),
+        output_tokens=(
+            output_tokens + thoughts
+            if output_tokens is not None and thoughts is not None
+            else None
+        ),
         cache_read_tokens=cache_read,
         total_tokens=total,
     )