allegro/electra-archon
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(state): Implement State Schema for SEED Architecture (Issue #3)

Browse Source 
- Add JSON schema at schemas/state.json with full validation rules
- Implement State, StateType, StateMetadata dataclasses in models/state.py
- Support immutable state objects with versioning and TTL
- Include serialization/deserialization (JSON and dict)
- Add next_version() for state history chaining
- Comprehensive test suite with 19 tests (100% pass)
- Full documentation at docs/state-schema.md

Features:
- UUID validation for all ID fields
- Optimistic locking via version numbers
- Flexible payload structure
- Metadata support (source, provenance, tags)
- Expiration checking via TTL
- Complete type hints

Closes Issue #3
This commit is contained in:
Allegro
2026-04-02 19:59:35 +00:00
parent 286eb0050f
commit bcf8c31270
8 changed files with 818 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
README.md
View File

@@ -1,3 +1,65 @@
# electra-archon
# Electra Archon
Electra Archon - SEED Architecture Implementation
SEED Architecture Implementation for the Electra Archon wizard.
## SEED Architecture
SEED stands for **S**tate-**E**vent-**E**ntity-**D**omain - a modular architecture for building resilient, auditable systems.
### Components
1. **State** - Immutable snapshots of entity state with versioning
2. **Event** - Pub/sub system for inter-component communication
3. **Entity** - Core domain objects with identity
4. **Domain** - Business logic and rules
## Project Structure
```
electra-archon/
├── schemas/ # JSON schemas for data validation
│ └── state.json # State schema definition
├── models/ # Python data models
│ └── state.py # State dataclass implementation
├── docs/ # Documentation
│ └── state-schema.md
├── tests/ # Test suite
│ └── test_state.py
└── pytest.ini # Test configuration
```
## Quick Start
```python
from models.state import State, StateType, StateMetadata
import uuid
# Create a state
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={"status": "running"},
metadata=StateMetadata(source="electra", tags=["seed"])
)
# Serialize
json_str = state.to_json()
```
## Running Tests
```bash
pytest tests/ -v
```
## Backlog
See [Issues](http://143.198.27.163:3000/allegro/electra-archon/issues) for current backlog:
- Issue #3: Design Electra State Schema for SEED Architecture ✅
- Issue #4: Implement Event Bus for Inter-Archon Communication
- Issue #5: Create Entity Resolution Service
## License
MIT

153
docs/state-schema.md Normal file
View File

@@ -0,0 +1,153 @@
# SEED State Schema Documentation
## Overview
The State Schema defines the structure for persisting and managing state within the SEED (State-Event-Entity-Domain) architecture. Each state record represents a snapshot of an entity at a specific point in time.
## Schema Design
### Core Fields
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `id` | UUID | Yes | Unique identifier for this state record |
| `entity_id` | UUID | Yes | Reference to the entity this state belongs to |
| `state_type` | Enum | Yes | Classification: `active`, `inactive`, `pending`, `archived`, `deleted` |
| `payload` | Object | Yes | Flexible JSON containing state-specific data |
| `timestamp` | ISO 8601 | Yes | When this state was recorded |
| `version` | Integer | Yes | Version number for optimistic locking (>= 1) |
| `metadata` | Object | No | Additional metadata (source, provenance, tags) |
| `previous_state_id` | UUID | No | Reference to previous state (for history chain) |
| `ttl` | Integer | No | Time-to-live in seconds |
### State Types
- **active**: Entity is currently active and operational
- **inactive**: Entity is temporarily inactive
- **pending**: Entity is in a pending/waiting state
- **archived**: Entity has been archived
- **deleted**: Entity has been marked for deletion
### Metadata Structure
```json
{
"source": "service-name",
"provenance": "trace-id",
"created_by": "user-or-service",
"tags": ["tag1", "tag2"]
}
```
## Python Model
### Basic Usage
```python
from models.state import State, StateType, StateMetadata
import uuid
# Create a new state
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={"status": "running", "progress": 75},
metadata=StateMetadata(
source="electra-archon",
created_by="electra",
tags=["seed", "priority"]
)
)
# Serialize to JSON
json_str = state.to_json(indent=2)
# Deserialize from JSON
restored_state = State.from_json(json_str)
```
### Versioning
```python
# Create next version of a state
next_state = state.next_version({"status": "running", "progress": 90})
print(next_state.version) # 2
print(next_state.previous_state_id) # Original state ID
```
### Validation
The model validates:
- UUID format for all ID fields
- Version is >= 1
- TTL is non-negative if specified
- State type is valid enum value
## Indexes
For optimal performance, the following indexes are recommended:
```sql
CREATE INDEX idx_state_entity_id ON states(entity_id);
CREATE INDEX idx_state_timestamp ON states(timestamp);
CREATE INDEX idx_state_type ON states(state_type);
CREATE INDEX idx_state_entity_version ON states(entity_id, version);
```
## JSON Schema
The full JSON Schema is available at `schemas/state.json`. It can be used for:
- API request/response validation
- Documentation generation
- Client code generation
## Example State Object
```json
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"entity_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
"state_type": "active",
"payload": {
"status": "running",
"progress": 75,
"details": {
"last_action": "processing",
"queue_position": 3
}
},
"timestamp": "2026-04-02T19:30:00Z",
"version": 3,
"metadata": {
"source": "electra-archon",
"provenance": "trace-abc123",
"created_by": "electra",
"tags": ["seed", "priority"]
},
"previous_state_id": "550e8400-e29b-41d4-a716-446655440001",
"ttl": null
}
```
## Testing
Run the test suite:
```bash
pytest tests/test_state.py -v
```
## Integration with SEED Architecture
The State component works with:
- **Entity**: States reference entities via `entity_id`
- **Event**: State changes can emit events
- **Domain**: Business logic determines state transitions
## Future Enhancements
- [ ] Compression for large payloads
- [ ] Encryption for sensitive data
- [ ] State transition validation rules
- [ ] Bulk state operations
- [ ] State snapshot/restore functionality

BIN
models/__pycache__/state.cpython-312.pyc Normal file
View File

Binary file not shown.

233
models/state.py Normal file
View File

@@ -0,0 +1,233 @@
"""
State Model for SEED Architecture
This module defines the State dataclass and related utilities for the
State component of the SEED (State-Event-Entity-Domain) architecture.
"""
from __future__ import annotations
import json
import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum, auto
from typing import Any, Optional, Dict, List
from pathlib import Path
class StateType(Enum):
"""Enumeration of valid state types."""
ACTIVE = "active"
INACTIVE = "inactive"
PENDING = "pending"
ARCHIVED = "archived"
DELETED = "deleted"
@dataclass(frozen=True)
class StateMetadata:
"""Metadata associated with a state record."""
source: str = "unknown"
provenance: Optional[str] = None
created_by: Optional[str] = None
tags: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
"""Convert metadata to dictionary."""
return {
"source": self.source,
"provenance": self.provenance,
"created_by": self.created_by,
"tags": self.tags
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> StateMetadata:
"""Create metadata from dictionary."""
return cls(
source=data.get("source", "unknown"),
provenance=data.get("provenance"),
created_by=data.get("created_by"),
tags=data.get("tags", [])
)
@dataclass(frozen=True)
class State:
"""
Immutable State record for SEED architecture.
Attributes:
id: Unique identifier for this state record (UUID v4)
entity_id: Reference to the entity this state belongs to
state_type: The type/category of this state
payload: Flexible JSON payload containing state-specific data
timestamp: ISO 8601 timestamp when this state was recorded
version: Version number for optimistic locking
metadata: Additional metadata about this state record
previous_state_id: Reference to the previous state (None if first)
ttl: Time-to-live in seconds (None for no expiration)
"""
id: str
entity_id: str
state_type: StateType
payload: Dict[str, Any]
timestamp: datetime
version: int
metadata: StateMetadata = field(default_factory=StateMetadata)
previous_state_id: Optional[str] = None
ttl: Optional[int] = None
def __post_init__(self):
"""Validate state after initialization."""
# Validate UUIDs
try:
uuid.UUID(self.id)
uuid.UUID(self.entity_id)
if self.previous_state_id:
uuid.UUID(self.previous_state_id)
except ValueError as e:
raise ValueError(f"Invalid UUID format: {e}")
# Validate version
if self.version < 1:
raise ValueError("Version must be >= 1")
# Validate TTL
if self.ttl is not None and self.ttl < 0:
raise ValueError("TTL must be non-negative")
def to_dict(self) -> Dict[str, Any]:
"""Convert state to dictionary representation."""
return {
"id": self.id,
"entity_id": self.entity_id,
"state_type": self.state_type.value,
"payload": self.payload,
"timestamp": self.timestamp.isoformat(),
"version": self.version,
"metadata": self.metadata.to_dict(),
"previous_state_id": self.previous_state_id,
"ttl": self.ttl
}
def to_json(self, indent: Optional[int] = None) -> str:
"""Serialize state to JSON string."""
return json.dumps(self.to_dict(), indent=indent, default=str)
@classmethod
def create(
cls,
entity_id: str,
state_type: StateType,
payload: Dict[str, Any],
version: int = 1,
metadata: Optional[StateMetadata] = None,
previous_state_id: Optional[str] = None,
ttl: Optional[int] = None
) -> State:
"""
Factory method to create a new State with auto-generated ID and timestamp.
Args:
entity_id: The entity this state belongs to
state_type: Type of state
payload: State data payload
version: Version number (default: 1)
metadata: Optional metadata
previous_state_id: Link to previous state
ttl: Optional time-to-live in seconds
Returns:
New State instance
"""
return cls(
id=str(uuid.uuid4()),
entity_id=entity_id,
state_type=state_type,
payload=payload,
timestamp=datetime.now(timezone.utc),
version=version,
metadata=metadata or StateMetadata(),
previous_state_id=previous_state_id,
ttl=ttl
)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> State:
"""Create State from dictionary representation."""
return cls(
id=data["id"],
entity_id=data["entity_id"],
state_type=StateType(data["state_type"]),
payload=data["payload"],
timestamp=datetime.fromisoformat(data["timestamp"].replace('Z', '+00:00')),
version=data["version"],
metadata=StateMetadata.from_dict(data.get("metadata", {})),
previous_state_id=data.get("previous_state_id"),
ttl=data.get("ttl")
)
@classmethod
def from_json(cls, json_str: str) -> State:
"""Deserialize State from JSON string."""
return cls.from_dict(json.loads(json_str))
def is_expired(self) -> bool:
"""Check if this state has expired based on TTL."""
if self.ttl is None:
return False
age = (datetime.now(timezone.utc) - self.timestamp).total_seconds()
return age > self.ttl
def next_version(self, new_payload: Dict[str, Any]) -> State:
"""
Create a new State representing the next version of this state.
Args:
new_payload: Updated payload for the new state
Returns:
New State instance with incremented version
"""
return State.create(
entity_id=self.entity_id,
state_type=self.state_type,
payload=new_payload,
version=self.version + 1,
metadata=self.metadata,
previous_state_id=self.id,
ttl=self.ttl
)
def load_schema() -> Dict[str, Any]:
"""Load the JSON schema for validation."""
schema_path = Path(__file__).parent.parent / "schemas" / "state.json"
with open(schema_path, 'r') as f:
return json.load(f)
# Example usage
if __name__ == "__main__":
# Create a sample state
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={"status": "running", "progress": 75},
metadata=StateMetadata(
source="electra-archon",
created_by="electra",
tags=["seed", "priority"]
)
)
print("Created State:")
print(state.to_json(indent=2))
print(f"\nIs expired: {state.is_expired()}")
# Create next version
next_state = state.next_version({"status": "running", "progress": 90})
print(f"\nNext version: {next_state.version}")
print(f"Previous state ID: {next_state.previous_state_id}")

6
pytest.ini Normal file
View File

@@ -0,0 +1,6 @@
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --tb=short

99
schemas/state.json Normal file
View File

@@ -0,0 +1,99 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"$id": "https://allegro.local/schemas/state.json",
"title": "SEED State Schema",
"description": "JSON Schema for State objects in the SEED Architecture",
"type": "object",
"required": ["id", "entity_id", "state_type", "payload", "timestamp", "version"],
"properties": {
"id": {
"type": "string",
"format": "uuid",
"description": "Unique identifier for this state record (UUID v4)"
},
"entity_id": {
"type": "string",
"format": "uuid",
"description": "Reference to the entity this state belongs to"
},
"state_type": {
"type": "string",
"enum": ["active", "inactive", "pending", "archived", "deleted"],
"description": "The type/category of this state"
},
"payload": {
"type": "object",
"description": "Flexible JSON payload containing state-specific data"
},
"timestamp": {
"type": "string",
"format": "date-time",
"description": "ISO 8601 timestamp when this state was recorded"
},
"version": {
"type": "integer",
"minimum": 1,
"description": "Version number for optimistic locking"
},
"metadata": {
"type": "object",
"description": "Additional metadata about this state record",
"properties": {
"source": {
"type": "string",
"description": "Source system or service that created this state"
},
"provenance": {
"type": "string",
"description": "Trace ID or provenance information"
},
"created_by": {
"type": "string",
"description": "User or service that created this state"
},
"tags": {
"type": "array",
"items": {
"type": "string"
},
"description": "Optional tags for categorization"
}
}
},
"previous_state_id": {
"type": ["string", "null"],
"format": "uuid",
"description": "Reference to the previous state record for this entity (null if first)"
},
"ttl": {
"type": ["integer", "null"],
"description": "Time-to-live in seconds (null for no expiration)"
}
},
"additionalProperties": false,
"examples": [
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"entity_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
"state_type": "active",
"payload": {
"status": "running",
"progress": 75,
"details": {
"last_action": "processing",
"queue_position": 3
}
},
"timestamp": "2026-04-02T19:30:00Z",
"version": 3,
"metadata": {
"source": "electra-archon",
"provenance": "trace-abc123",
"created_by": "electra",
"tags": ["seed", "priority"]
},
"previous_state_id": "550e8400-e29b-41d4-a716-446655440001",
"ttl": null
}
]
}

BIN
tests/__pycache__/test_state.cpython-312-pytest-9.0.2.pyc Normal file
View File

Binary file not shown.

263
tests/test_state.py Normal file
View File

@@ -0,0 +1,263 @@
"""
Tests for State Model and Schema Validation
"""
import json
import uuid
import pytest
from datetime import datetime, timezone
from pathlib import Path
import sys
# Add parent to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from models.state import State, StateType, StateMetadata, load_schema
class TestStateMetadata:
"""Tests for StateMetadata class."""
def test_default_creation(self):
"""Test creating metadata with defaults."""
meta = StateMetadata()
assert meta.source == "unknown"
assert meta.provenance is None
assert meta.created_by is None
assert meta.tags == []
def test_full_creation(self):
"""Test creating metadata with all fields."""
meta = StateMetadata(
source="test-source",
provenance="trace-123",
created_by="electra",
tags=["seed", "test"]
)
assert meta.source == "test-source"
assert meta.provenance == "trace-123"
assert meta.created_by == "electra"
assert meta.tags == ["seed", "test"]
def test_to_dict(self):
"""Test metadata serialization."""
meta = StateMetadata(source="test", tags=["a", "b"])
d = meta.to_dict()
assert d["source"] == "test"
assert d["tags"] == ["a", "b"]
assert d["provenance"] is None
def test_from_dict(self):
"""Test metadata deserialization."""
data = {"source": "src", "provenance": "prov", "created_by": "user", "tags": ["t"]}
meta = StateMetadata.from_dict(data)
assert meta.source == "src"
assert meta.provenance == "prov"
class TestState:
"""Tests for State class."""
def test_create_with_defaults(self):
"""Test State.create factory method."""
entity_id = str(uuid.uuid4())
state = State.create(
entity_id=entity_id,
state_type=StateType.ACTIVE,
payload={"key": "value"}
)
assert state.entity_id == entity_id
assert state.state_type == StateType.ACTIVE
assert state.payload == {"key": "value"}
assert state.version == 1
assert state.previous_state_id is None
assert state.ttl is None
# Verify UUID format
uuid.UUID(state.id) # Should not raise
def test_create_with_metadata(self):
"""Test State.create with metadata."""
meta = StateMetadata(source="test", tags=["important"])
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.PENDING,
payload={},
metadata=meta
)
assert state.metadata.source == "test"
assert state.metadata.tags == ["important"]
def test_invalid_uuid_raises(self):
"""Test that invalid UUIDs raise ValueError."""
with pytest.raises(ValueError, match="Invalid UUID"):
State(
id="not-a-uuid",
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={},
timestamp=datetime.now(timezone.utc),
version=1
)
def test_invalid_version_raises(self):
"""Test that version < 1 raises ValueError."""
with pytest.raises(ValueError, match="Version must be >= 1"):
State(
id=str(uuid.uuid4()),
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={},
timestamp=datetime.now(timezone.utc),
version=0
)
def test_negative_ttl_raises(self):
"""Test that negative TTL raises ValueError."""
with pytest.raises(ValueError, match="TTL must be non-negative"):
State(
id=str(uuid.uuid4()),
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={},
timestamp=datetime.now(timezone.utc),
version=1,
ttl=-1
)
def test_to_dict_serialization(self):
"""Test state to dictionary conversion."""
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={"count": 42}
)
d = state.to_dict()
assert d["id"] == state.id
assert d["entity_id"] == state.entity_id
assert d["state_type"] == "active"
assert d["payload"] == {"count": 42}
assert "timestamp" in d
assert d["version"] == 1
def test_to_json_serialization(self):
"""Test state to JSON conversion."""
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ARCHIVED,
payload={"archived": True}
)
json_str = state.to_json()
# Should be valid JSON
parsed = json.loads(json_str)
assert parsed["state_type"] == "archived"
assert parsed["payload"]["archived"] is True
def test_from_dict_deserialization(self):
"""Test state from dictionary conversion."""
entity_id = str(uuid.uuid4())
data = {
"id": str(uuid.uuid4()),
"entity_id": entity_id,
"state_type": "pending",
"payload": {"status": "waiting"},
"timestamp": "2026-04-02T12:00:00+00:00",
"version": 2,
"metadata": {"source": "test"},
"previous_state_id": str(uuid.uuid4()),
"ttl": 3600
}
state = State.from_dict(data)
assert state.entity_id == entity_id
assert state.state_type == StateType.PENDING
assert state.version == 2
assert state.ttl == 3600
def test_from_json_deserialization(self):
"""Test state from JSON conversion."""
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.DELETED,
payload={"deleted_at": "2026-04-01"}
)
json_str = state.to_json()
restored = State.from_json(json_str)
assert restored.id == state.id
assert restored.state_type == StateType.DELETED
assert restored.payload == state.payload
def test_is_expired_no_ttl(self):
"""Test that state without TTL never expires."""
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={}
)
assert not state.is_expired()
def test_is_expired_with_ttl(self):
"""Test TTL expiration logic."""
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={},
ttl=1 # 1 second TTL
)
assert not state.is_expired()
# Note: We can't easily test actual expiration without sleeping
def test_next_version(self):
"""Test creating next version of state."""
state = State.create(
entity_id=str(uuid.uuid4()),
state_type=StateType.ACTIVE,
payload={"count": 1},
version=5
)
next_state = state.next_version({"count": 2})
assert next_state.version == 6
assert next_state.previous_state_id == state.id
assert next_state.entity_id == state.entity_id
assert next_state.payload == {"count": 2}
assert next_state.state_type == state.state_type
def test_all_state_types(self):
"""Test all state type enum values."""
entity_id = str(uuid.uuid4())
for state_type in StateType:
state = State.create(
entity_id=entity_id,
state_type=state_type,
payload={}
)
assert state.state_type == state_type
assert state.to_dict()["state_type"] == state_type.value
class TestSchema:
"""Tests for JSON Schema."""
def test_schema_loads(self):
"""Test that schema file loads correctly."""
schema = load_schema()
assert schema["title"] == "SEED State Schema"
assert "properties" in schema
assert "id" in schema["required"]
assert "entity_id" in schema["required"]
def test_schema_has_examples(self):
"""Test that schema includes examples."""
schema = load_schema()
assert "examples" in schema
assert len(schema["examples"]) > 0
if __name__ == "__main__":
pytest.main([__file__, "-v"])