gunn.schemas package

Submodules

gunn.schemas.messages module

Pydantic models for data validation and JSON schema generation.

class gunn.schemas.messages.WorldState(**data)[ソース]

ベースクラス: BaseModel

Complete world state containing all entities and relationships.

パラメータ:
entities: dict[str, Any]
relationships: dict[str, list[str]]
spatial_index: dict[str, tuple[float, float, float]]
metadata: dict[str, Any]
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'json_schema_extra': {'examples': [{'entities': {'agent1': {'health': 100, 'name': 'Alice'}}, 'metadata': {'world_name': 'test_world'}, 'relationships': {'agent1': ['agent2']}, 'spatial_index': {'agent1': [10.0, 20.0, 0.0]}}]}, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_spatial_coordinates(v)[ソース]

Validate that spatial coordinates are 3D tuples of floats.

戻り値の型:

dict[str, tuple[float, float, float]]

パラメータ:

v (dict[str, tuple[float, float, float]])

class gunn.schemas.messages.View(**data)[ソース]

ベースクラス: BaseModel

Agent's filtered view of the world state.

パラメータ:
agent_id: str
view_seq: int
visible_entities: dict[str, Any]
visible_relationships: dict[str, list[str]]
context_digest: str
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'json_schema_extra': {'examples': [{'agent_id': 'agent_001', 'context_digest': 'sha256:abc123def456', 'view_seq': 42, 'visible_entities': {'entity1': {'name': 'Alice', 'type': 'player'}}, 'visible_relationships': {'entity1': ['entity2']}}]}, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.schemas.messages.EventLogEntry(**data)[ソース]

ベースクラス: BaseModel

Single entry in the event log with integrity checking.

パラメータ:
global_seq: int
sim_time: float
wall_time: float
effect: dict[str, Any]
source_metadata: dict[str, Any]
checksum: str
req_id: str
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'json_schema_extra': {'examples': [{'checksum': 'sha256:def789abc123', 'effect': {'global_seq': 1001, 'kind': 'Move', 'payload': {'x': 10, 'y': 20}, 'schema_version': '1.0.0', 'sim_time': 123.456, 'source_id': 'agent_001', 'uuid': '550e8400-e29b-41d4-a716-446655440000'}, 'global_seq': 1001, 'req_id': 'req_550e8400-e29b-41d4', 'sim_time': 123.456, 'source_metadata': {'adapter': 'unity', 'world_id': 'world_001'}, 'wall_time': 1640995200.0}]}, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_effect_structure(v)[ソース]

Validate that effect has required fields.

戻り値の型:

dict[str, Any]

パラメータ:

v (dict[str, Any])

gunn.schemas.types module

Core type definitions using TypedDict for performance and schema versioning.

class gunn.schemas.types.Intent[ソース]

ベースクラス: TypedDict

Intent submitted by an agent to perform an action.

This represents a request from an agent to perform a specific action in the world. Intents go through validation before being converted to Effects.

kind: Annotated[Literal['Speak', 'Move', 'Interact', 'Custom']]
payload: Annotated[dict[str, Any]]
context_seq: Annotated[int]
req_id: Annotated[str]
agent_id: Annotated[str]
priority: Annotated[int]
schema_version: Annotated[str]
class gunn.schemas.types.EffectDraft[ソース]

ベースクラス: TypedDict

External input for effects - Orchestrator fills in uuid, global_seq, sim_time.

This is the incomplete form of an Effect, used when external systems (like game engines) want to broadcast events. The Orchestrator completes it with timing and sequence info.

kind: Annotated[str]
payload: Annotated[dict[str, Any]]
source_id: Annotated[str]
schema_version: Annotated[str]
class gunn.schemas.types.Effect[ソース]

ベースクラス: TypedDict

Complete effect with all fields filled by Orchestrator.

uuid: Annotated[str]
kind: Annotated[str]
payload: Annotated[dict[str, Any]]
global_seq: Annotated[int]
sim_time: Annotated[float]
source_id: Annotated[str]
schema_version: Annotated[str]
class gunn.schemas.types.ObservationDelta[ソース]

ベースクラス: TypedDict

RFC6902 JSON Patch operations for incremental view updates.

view_seq: Annotated[int]
patches: Annotated[list[dict[str, Any]]]
context_digest: Annotated[str]
schema_version: Annotated[str]
class gunn.schemas.types.CancelToken(req_id, agent_id)[ソース]

ベースクラス: object

Token for tracking and cancelling agent generation operations.

This class provides a thread-safe way to signal cancellation of long-running operations like LLM generation, with reason tracking for debugging.

パラメータ:
property cancelled: bool

Check if the token has been cancelled.

cancel(reason)[ソース]

Cancel the token with a specific reason.

戻り値の型:

None

パラメータ:

reason (str)

Args:

reason: Human-readable reason for cancellation

Raises:

ValueError: If reason is empty

async wait_cancelled()[ソース]

Wait until token is cancelled.

This method will block until cancel() is called on this token.

戻り値の型:

None

Module contents

Data models and type definitions for the gunn simulation core.

class gunn.schemas.CancelToken(req_id, agent_id)[ソース]

ベースクラス: object

Token for tracking and cancelling agent generation operations.

This class provides a thread-safe way to signal cancellation of long-running operations like LLM generation, with reason tracking for debugging.

パラメータ:
cancel(reason)[ソース]

Cancel the token with a specific reason.

戻り値の型:

None

パラメータ:

reason (str)

Args:

reason: Human-readable reason for cancellation

Raises:

ValueError: If reason is empty

property cancelled: bool

Check if the token has been cancelled.

async wait_cancelled()[ソース]

Wait until token is cancelled.

This method will block until cancel() is called on this token.

戻り値の型:

None

class gunn.schemas.Effect[ソース]

ベースクラス: TypedDict

Complete effect with all fields filled by Orchestrator.

uuid: Annotated[str]
kind: Annotated[str]
payload: Annotated[dict[str, Any]]
global_seq: Annotated[int]
sim_time: Annotated[float]
source_id: Annotated[str]
schema_version: Annotated[str]
class gunn.schemas.EffectDraft[ソース]

ベースクラス: TypedDict

External input for effects - Orchestrator fills in uuid, global_seq, sim_time.

This is the incomplete form of an Effect, used when external systems (like game engines) want to broadcast events. The Orchestrator completes it with timing and sequence info.

kind: Annotated[str]
payload: Annotated[dict[str, Any]]
source_id: Annotated[str]
schema_version: Annotated[str]
class gunn.schemas.EventLogEntry(**data)[ソース]

ベースクラス: BaseModel

Single entry in the event log with integrity checking.

パラメータ:
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'json_schema_extra': {'examples': [{'checksum': 'sha256:def789abc123', 'effect': {'global_seq': 1001, 'kind': 'Move', 'payload': {'x': 10, 'y': 20}, 'schema_version': '1.0.0', 'sim_time': 123.456, 'source_id': 'agent_001', 'uuid': '550e8400-e29b-41d4-a716-446655440000'}, 'global_seq': 1001, 'req_id': 'req_550e8400-e29b-41d4', 'sim_time': 123.456, 'source_metadata': {'adapter': 'unity', 'world_id': 'world_001'}, 'wall_time': 1640995200.0}]}, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_effect_structure(v)[ソース]

Validate that effect has required fields.

戻り値の型:

dict[str, Any]

パラメータ:

v (dict[str, Any])

global_seq: int
sim_time: float
wall_time: float
effect: dict[str, Any]
source_metadata: dict[str, Any]
checksum: str
req_id: str
class gunn.schemas.Intent[ソース]

ベースクラス: TypedDict

Intent submitted by an agent to perform an action.

This represents a request from an agent to perform a specific action in the world. Intents go through validation before being converted to Effects.

kind: Annotated[Literal['Speak', 'Move', 'Interact', 'Custom']]
payload: Annotated[dict[str, Any]]
context_seq: Annotated[int]
req_id: Annotated[str]
agent_id: Annotated[str]
priority: Annotated[int]
schema_version: Annotated[str]
class gunn.schemas.ObservationDelta[ソース]

ベースクラス: TypedDict

RFC6902 JSON Patch operations for incremental view updates.

view_seq: Annotated[int]
patches: Annotated[list[dict[str, Any]]]
context_digest: Annotated[str]
schema_version: Annotated[str]
class gunn.schemas.View(**data)[ソース]

ベースクラス: BaseModel

Agent's filtered view of the world state.

パラメータ:
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'json_schema_extra': {'examples': [{'agent_id': 'agent_001', 'context_digest': 'sha256:abc123def456', 'view_seq': 42, 'visible_entities': {'entity1': {'name': 'Alice', 'type': 'player'}}, 'visible_relationships': {'entity1': ['entity2']}}]}, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

agent_id: str
view_seq: int
visible_entities: dict[str, Any]
visible_relationships: dict[str, list[str]]
context_digest: str
class gunn.schemas.WorldState(**data)[ソース]

ベースクラス: BaseModel

Complete world state containing all entities and relationships.

パラメータ:
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'json_schema_extra': {'examples': [{'entities': {'agent1': {'health': 100, 'name': 'Alice'}}, 'metadata': {'world_name': 'test_world'}, 'relationships': {'agent1': ['agent2']}, 'spatial_index': {'agent1': [10.0, 20.0, 0.0]}}]}, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_spatial_coordinates(v)[ソース]

Validate that spatial coordinates are 3D tuples of floats.

戻り値の型:

dict[str, tuple[float, float, float]]

パラメータ:

v (dict[str, tuple[float, float, float]])

entities: dict[str, Any]
relationships: dict[str, list[str]]
spatial_index: dict[str, tuple[float, float, float]]
metadata: dict[str, Any]