feat(models): AiderEventIn + AiderBatchIn pydantic schemas

- Implement aider-watch v2 event schema with 7 event types
- Enforce timezone-aware timestamps via field_validator
- Batch schema supports up to 50 events per request
- Frozen + forbid extra fields (defensive engineering)
- Fix broken src.* imports in models package (incident.py, __init__.py)

Task A3 complete: 7/7 tests passing
This commit is contained in:
Your Name
2026-04-20 04:06:26 +08:00
parent 0db4534133
commit 5daae76147
4 changed files with 107 additions and 4 deletions

View File

@@ -10,7 +10,7 @@ AWOOOI Models Package
"""
# Approval Models (Phase 2)
from src.models.approval import (
from apps.api.src.models.approval import (
ApprovalRequest,
ApprovalRequestCreate,
ApprovalRequestResponse,
@@ -28,7 +28,7 @@ from src.models.approval import (
)
# Incident Models (Phase 6 - 認知覺醒)
from src.models.incident import (
from apps.api.src.models.incident import (
AIDecisionChain,
Incident,
IncidentCreate,
@@ -41,7 +41,7 @@ from src.models.incident import (
)
# NVIDIA Models (ADR-036 - Nemotron Tool Calling)
from src.models.nvidia import (
from apps.api.src.models.nvidia import (
NvidiaProviderResult,
NvidiaResponse,
NvidiaUsage,
@@ -50,6 +50,13 @@ from src.models.nvidia import (
ToolDefinition,
)
# Aider Models (aider-watch v2)
from apps.api.src.models.aider import (
AiderBatchIn,
AiderEventIn,
EventType,
)
__all__ = [
# Approval
"ApprovalRequest",
@@ -83,4 +90,8 @@ __all__ = [
"ToolCall",
"ToolCallValidationResult",
"ToolDefinition",
# Aider (aider-watch v2)
"AiderBatchIn",
"AiderEventIn",
"EventType",
]

View File

@@ -0,0 +1,35 @@
"""aider-watch event pydantic schemas。由 Mac client POST 送來。"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
EventType = Literal[
"session_start", "file_edit", "error", "commit",
"silent_timeout", "session_end", "raw",
]
class AiderEventIn(BaseModel):
"""單筆 aider event 輸入 schema。"""
model_config = ConfigDict(frozen=True, extra="forbid")
ts: datetime
session_id: str = Field(min_length=1, max_length=64)
host: str = Field(default="ogt-mac", max_length=64)
type: EventType
payload: dict[str, Any] = Field(default_factory=dict)
@field_validator("ts")
@classmethod
def _ensure_tz(cls, v: datetime) -> datetime:
if v.tzinfo is None:
raise ValueError("ts must be timezone-aware")
return v
class AiderBatchIn(BaseModel):
"""批次 event 輸入(最多 50 筆/次)。"""
model_config = ConfigDict(frozen=True, extra="forbid")
events: list[AiderEventIn] = Field(min_length=1, max_length=50)

View File

@@ -28,7 +28,7 @@ from uuid import UUID, uuid4
from pydantic import BaseModel, Field, field_validator
# 復用現有模型 (避免重複定義)
from src.models.approval import BlastRadius
from apps.api.src.models.approval import BlastRadius
# =============================================================================
# Incident 專用 Enums

View File

@@ -0,0 +1,57 @@
# apps/api/tests/test_aider_event_models.py | 2026-04-20 @ Asia/Taipei
from datetime import datetime, timezone, timedelta
import pytest
from pydantic import ValidationError
from apps.api.src.models.aider import AiderEventIn, AiderBatchIn
TAIPEI = timezone(timedelta(hours=8))
def _base(t="session_start", payload=None):
return {
"ts": datetime(2026, 4, 20, 10, 0, tzinfo=TAIPEI).isoformat(),
"session_id": "01J7XYZABC",
"host": "ogt-mac",
"type": t,
"payload": payload or {},
}
def test_accepts_all_7_types():
for t in ("session_start", "file_edit", "error", "commit",
"silent_timeout", "session_end", "raw"):
ev = AiderEventIn(**_base(t=t))
assert ev.type == t
def test_rejects_unknown_type():
with pytest.raises(ValidationError):
AiderEventIn(**_base(t="bogus"))
def test_rejects_missing_session_id():
d = _base(); del d["session_id"]
with pytest.raises(ValidationError):
AiderEventIn(**d)
def test_rejects_missing_ts():
d = _base(); del d["ts"]
with pytest.raises(ValidationError):
AiderEventIn(**d)
def test_batch_max_50():
with pytest.raises(ValidationError):
AiderBatchIn(events=[_base() for _ in range(51)])
def test_batch_accepts_1():
b = AiderBatchIn(events=[_base()])
assert len(b.events) == 1
def test_host_default():
d = _base(); del d["host"]
ev = AiderEventIn(**d)
assert ev.host == "ogt-mac"