58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""Typed contracts for the Agent queue protocol."""
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
class AgentCommand(BaseModel):
|
|
command_id: str
|
|
command_type: str
|
|
request_payload: dict[str, Any]
|
|
created_at: datetime
|
|
|
|
|
|
class AgentCommandListResponse(BaseModel):
|
|
ok: Literal[True] = True
|
|
commands: list[AgentCommand]
|
|
|
|
|
|
class MinerControlResultPayload(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
schema_version: Literal[1]
|
|
command_accepted: bool
|
|
verification_attempted: bool
|
|
verification_status: Literal[
|
|
"confirmed", "verification_failed", "unconfirmed", "unavailable"
|
|
]
|
|
observed_service_state: None = None
|
|
observed_performance_mode: Literal["low", "normal", "high"] | None = None
|
|
failure_code: str | None = Field(
|
|
default=None, max_length=80, pattern=r"^[a-z0-9_]+$"
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_truthfulness(self) -> "MinerControlResultPayload":
|
|
if self.verification_status == "confirmed" and not self.verification_attempted:
|
|
raise ValueError("confirmed results require verification_attempted")
|
|
if not self.command_accepted and not self.failure_code:
|
|
raise ValueError("rejected commands require a failure_code")
|
|
return self
|
|
|
|
|
|
class AgentCommandResultRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
command_id: str = Field(min_length=1, max_length=128)
|
|
status: Literal["completed", "failed"]
|
|
result_payload: dict[str, Any] = Field(default_factory=dict)
|
|
error_message: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class AgentCommandResultResponse(BaseModel):
|
|
ok: Literal[True] = True
|
|
command_id: str
|
|
status: Literal["completed", "failed"]
|