"""Dataclass models for AWM resources and activities."""
from dataclasses import dataclass
from datetime import datetime
from typing import Any
from chariot.awm._utils import convert_to_datetime
__all__ = [
"Activity",
"ConfigurationComponent",
"ConfigurationComponentVersion",
"ConfigurationComponentVersionSummary",
"ConfigurationStage",
"ConfigurationType",
"COMPONENT_TYPE_AGENT",
"COMPONENT_TYPE_EVALUATOR",
"COMPONENT_TYPE_MCP_CONNECTION",
"COMPONENT_TYPE_SHARED",
"COMPONENT_TYPE_TOOL_PROFILE",
"COMPONENT_TYPE_WORKFLOW",
"ResolvedConfiguration",
"STAGE_DEVELOPMENT",
"STAGE_PRODUCTION",
"STAGE_STAGING",
"Schedule",
"SessionDetails",
"Workflow",
"WorkflowResourceDetails",
"WorkflowPublicSettingRef",
"WorkflowPublicSettingValue",
"PodDetails",
]
ConfigurationType = str
ConfigurationStage = str
COMPONENT_TYPE_WORKFLOW = "workflow"
COMPONENT_TYPE_AGENT = "agent"
COMPONENT_TYPE_TOOL_PROFILE = "tool-profile"
COMPONENT_TYPE_MCP_CONNECTION = "mcp-connection"
COMPONENT_TYPE_EVALUATOR = "evaluator"
COMPONENT_TYPE_SHARED = "shared"
STAGE_DEVELOPMENT = "development"
STAGE_STAGING = "staging"
STAGE_PRODUCTION = "production"
class Base:
def __post_init__(self):
pass
[docs]
@dataclass
class WorkflowPublicSettingRef(Base):
"""A reference to a public setting, with optional schema."""
ref: str
schema: Any | None = None
[docs]
@dataclass
class WorkflowPublicSettingValue(Base):
"""A public setting value. The schema (if one exists), should be treated as read-only and is used for documentation purposes."""
value: Any
schema: Any | None = None
[docs]
@dataclass
class Workflow(Base):
"""An instance of a workflow."""
id: str
name: str
description: str
project_id: str
paused: bool
config: dict[str, Any] | None
public_settings: dict[str, WorkflowPublicSettingRef] | None
created_at: datetime
created_by: str
updated_at: datetime
updated_by: str
deleted_at: datetime | None
deleted_by: str | None
def __post_init__(self):
"""Normalize datetime/public-settings fields after construction."""
super().__post_init__()
self.created_at = convert_to_datetime(self.created_at)
self.updated_at = convert_to_datetime(self.updated_at)
self.deleted_at = (
convert_to_datetime(self.deleted_at) if self.deleted_at is not None else None
)
if self.public_settings is not None:
self.public_settings = {
k: WorkflowPublicSettingRef(
ref=str(v.get("ref")), schema=v.get("schema", v.get("var_schema"))
)
if isinstance(v, dict)
else v
for k, v in self.public_settings.items()
}
[docs]
@dataclass
class Activity(Base):
"""An activity event within a workflow session."""
id: str
workflow_id: str
type: str
body: Any
session_id: str
final: bool
created_at: datetime
from_agent: str = ""
to_agent: str = ""
user_id: str = ""
processed_at: datetime | None = None
session_context_size: int | None = None
def __post_init__(self):
"""Normalize datetime fields after construction."""
super().__post_init__()
self.created_at = convert_to_datetime(self.created_at)
self.processed_at = (
convert_to_datetime(self.processed_at) if self.processed_at is not None else None
)
[docs]
@dataclass
class Schedule(Base):
"""A scheduled task for a workflow."""
id: str
workflow_id: str
name: str
to_agent: str
prompt: str
active: bool
schedule: str
created_by: str
created_at: datetime
expires_at: datetime | None = None
expires_in: str | None = None
def __post_init__(self):
"""Normalize datetime fields after construction."""
super().__post_init__()
self.created_at = convert_to_datetime(self.created_at)
self.expires_at = (
convert_to_datetime(self.expires_at) if self.expires_at is not None else None
)
[docs]
@dataclass
class SessionDetails(Base):
"""Summary of a workflow session."""
session_id: str
status: str
agents: list[str]
user_ids: list[str]
total_activity_count: int
user_activity_count: int
tool_activity_count: int
created_at: datetime
updated_at: datetime
schedule_id: str | None = None
total_request_duration_micros: int | None = None
def __post_init__(self):
"""Normalize datetime fields after construction."""
super().__post_init__()
self.created_at = convert_to_datetime(self.created_at)
self.updated_at = convert_to_datetime(self.updated_at)
[docs]
@dataclass
class PodDetails(Base):
"""Pod details."""
name: str
status: str
[docs]
@dataclass
class ConfigurationComponent(Base):
"""A configuration component."""
id: str
project_id: str
component_type: str
name: str
key: str
deleted_at: datetime | None = None
def __post_init__(self):
"""Normalize datetime fields after construction."""
super().__post_init__()
self.deleted_at = (
convert_to_datetime(self.deleted_at) if self.deleted_at is not None else None
)
[docs]
@dataclass
class ConfigurationComponentVersionSummary(Base):
"""Summary metadata for a configuration component version."""
id: str
component_id: str
component_type: str
project_id: str
name: str
key: str
description: str
version: int
created_at: datetime
created_by: str
version_changelog: str | None = None
stages: list[str] | None = None
deleted_at: datetime | None = None
deleted_by: str | None = None
def __post_init__(self):
"""Normalize datetime fields after construction."""
super().__post_init__()
self.created_at = convert_to_datetime(self.created_at)
self.deleted_at = (
convert_to_datetime(self.deleted_at) if self.deleted_at is not None else None
)
[docs]
@dataclass
class ConfigurationComponentVersion(Base):
"""A configuration component version."""
id: str
component_id: str
component_type: str
project_id: str
name: str
key: str
description: str
version: int
created_at: datetime
created_by: str
config: dict[str, Any] | None = None
raw_config: list[int] | None = None
version_changelog: str | None = None
stages: list[str] | None = None
deleted_at: datetime | None = None
deleted_by: str | None = None
def __post_init__(self):
"""Normalize datetime fields after construction."""
super().__post_init__()
self.created_at = convert_to_datetime(self.created_at)
self.deleted_at = (
convert_to_datetime(self.deleted_at) if self.deleted_at is not None else None
)
[docs]
@dataclass
class ResolvedConfiguration(Base):
"""A configuration component version with optional resolved config."""
data: ConfigurationComponentVersion
resolved: dict[str, Any] | None = None
[docs]
@dataclass
class WorkflowResourceDetails(Base):
"""Resources of a workflow. Valid is false if any resources are missing, and any attempts to use this workflow will likely fail."""
valid: bool
deployment: str
pods: list[PodDetails]