stroblmeandClaude Fable 5
af3ba51571
A second bus subscriber folds executions, errors, timings and queue lag into per-minute rollups, keeps failures with their traceback and an audit trail of who published what, and records one row per cascade — manual runs and previews included, under an id of their own that writes no idempotency markers. Read back through /observability/*, which always answers 200 so a degraded engine still renders its own health screen. Also fixes two things found on the way: node-health alerts read `status` where the engine publishes `health`, so a device dropping never alerted anyone, and the Redis queue reported `parked: 0` whatever was held. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt
282 lines
8.7 KiB
Python
282 lines
8.7 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from pydantic import EmailStr
|
|
from sqlalchemy import JSON, Column, DateTime
|
|
from sqlmodel import Field, SQLModel
|
|
|
|
|
|
def get_datetime_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
# Shared properties
|
|
class UserBase(SQLModel):
|
|
email: EmailStr = Field(unique=True, index=True, max_length=255)
|
|
is_active: bool = True
|
|
is_superuser: bool = False
|
|
full_name: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
# Properties to receive via API on creation
|
|
class UserCreate(UserBase):
|
|
password: str = Field(min_length=8, max_length=128)
|
|
|
|
|
|
class UserRegister(SQLModel):
|
|
email: EmailStr = Field(max_length=255)
|
|
password: str = Field(min_length=8, max_length=128)
|
|
full_name: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
# Properties to receive via API on update, all are optional
|
|
class UserUpdate(UserBase):
|
|
email: EmailStr | None = Field(default=None, max_length=255) # type: ignore
|
|
password: str | None = Field(default=None, min_length=8, max_length=128)
|
|
|
|
|
|
class UserUpdateMe(SQLModel):
|
|
full_name: str | None = Field(default=None, max_length=255)
|
|
email: EmailStr | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class UpdatePassword(SQLModel):
|
|
current_password: str = Field(min_length=8, max_length=128)
|
|
new_password: str = Field(min_length=8, max_length=128)
|
|
|
|
|
|
# Database model, database table inferred from class name
|
|
class User(UserBase, table=True):
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
hashed_password: str
|
|
created_at: datetime | None = Field(
|
|
default_factory=get_datetime_utc,
|
|
sa_type=DateTime(timezone=True), # type: ignore
|
|
)
|
|
|
|
|
|
# Properties to return via API, id is always required
|
|
class UserPublic(UserBase):
|
|
id: uuid.UUID
|
|
created_at: datetime | None = None
|
|
|
|
|
|
class UsersPublic(SQLModel):
|
|
data: list[UserPublic]
|
|
count: int
|
|
|
|
|
|
# Generic message
|
|
class Message(SQLModel):
|
|
message: str
|
|
|
|
|
|
# JSON payload containing access token
|
|
class Token(SQLModel):
|
|
access_token: str
|
|
token_type: str = "bearer"
|
|
|
|
|
|
# Contents of JWT token
|
|
class TokenPayload(SQLModel):
|
|
sub: str | None = None
|
|
|
|
|
|
class NewPassword(SQLModel):
|
|
token: str
|
|
new_password: str = Field(min_length=8, max_length=128)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# OAuth 2.1, for the MCP endpoint
|
|
#
|
|
# Agents cannot be handed a password, so they get their own authorization flow:
|
|
# the client registers itself, a human approves it in the browser, and the code
|
|
# that comes back is exchanged for a token. Only the hash of a code or a refresh
|
|
# token is stored, so a copy of this table is not a set of credentials.
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class OAuthClient(SQLModel, table=True):
|
|
"""A registered agent. Registration alone grants nothing."""
|
|
|
|
__tablename__ = "oauth_client"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
client_name: str = Field(max_length=128)
|
|
redirect_uris: list[str] = Field(sa_column=Column(JSON), default_factory=list)
|
|
created_at: datetime = Field(
|
|
default_factory=lambda: datetime.now(timezone.utc), nullable=False
|
|
)
|
|
|
|
|
|
class OAuthAuthorizationCode(SQLModel, table=True):
|
|
"""One approved authorization, waiting to be exchanged for a token."""
|
|
|
|
__tablename__ = "oauth_authorization_code"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
code_hash: str = Field(max_length=64, unique=True, index=True)
|
|
client_id: uuid.UUID = Field(
|
|
foreign_key="oauth_client.id", nullable=False, ondelete="CASCADE"
|
|
)
|
|
user_id: uuid.UUID = Field(
|
|
foreign_key="user.id", nullable=False, ondelete="CASCADE"
|
|
)
|
|
redirect_uri: str = Field(max_length=2048)
|
|
code_challenge: str = Field(max_length=128)
|
|
resource: str | None = Field(default=None, max_length=2048)
|
|
expires_at: datetime
|
|
used_at: datetime | None = None
|
|
#: The refresh token this code produced, so replaying the code can revoke it.
|
|
refresh_token_id: uuid.UUID | None = None
|
|
created_at: datetime = Field(
|
|
default_factory=lambda: datetime.now(timezone.utc), nullable=False
|
|
)
|
|
|
|
|
|
class OAuthRefreshToken(SQLModel, table=True):
|
|
"""A rotating refresh token, one family per authorization."""
|
|
|
|
__tablename__ = "oauth_refresh_token"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
token_hash: str = Field(max_length=64, unique=True, index=True)
|
|
client_id: uuid.UUID = Field(
|
|
foreign_key="oauth_client.id", nullable=False, ondelete="CASCADE"
|
|
)
|
|
user_id: uuid.UUID = Field(
|
|
foreign_key="user.id", nullable=False, ondelete="CASCADE"
|
|
)
|
|
#: Every token rotated out of one authorization shares this, so reusing an
|
|
#: old one can revoke the whole line rather than just itself.
|
|
family_id: uuid.UUID
|
|
expires_at: datetime
|
|
revoked: bool = False
|
|
created_at: datetime = Field(
|
|
default_factory=lambda: datetime.now(timezone.utc), nullable=False
|
|
)
|
|
|
|
|
|
class OAuthClientRegister(SQLModel):
|
|
"""RFC 7591 dynamic client registration request."""
|
|
|
|
client_name: str = Field(default="MCP client", max_length=128)
|
|
redirect_uris: list[str]
|
|
grant_types: list[str] | None = None
|
|
response_types: list[str] | None = None
|
|
token_endpoint_auth_method: str | None = None
|
|
|
|
|
|
class OAuthClientInfo(SQLModel):
|
|
client_id: str
|
|
client_name: str
|
|
redirect_uris: list[str]
|
|
client_id_issued_at: int
|
|
grant_types: list[str] = ["authorization_code", "refresh_token"]
|
|
response_types: list[str] = ["code"]
|
|
token_endpoint_auth_method: str = "none"
|
|
|
|
|
|
class OAuthAuthorizeInfo(SQLModel):
|
|
"""What the consent page shows, all of it validated server-side."""
|
|
|
|
client_name: str
|
|
redirect_uri: str
|
|
scope: str
|
|
|
|
|
|
class OAuthAuthorizeRequest(SQLModel):
|
|
client_id: str
|
|
redirect_uri: str
|
|
code_challenge: str
|
|
code_challenge_method: str = "S256"
|
|
state: str | None = None
|
|
resource: str | None = None
|
|
scope: str | None = None
|
|
|
|
|
|
class OAuthAuthorizeResponse(SQLModel):
|
|
redirect_url: str
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Observability
|
|
#
|
|
# The engine's own history: what ran, how long it took, and what went wrong.
|
|
# Rolled up per minute rather than kept per execution — a node firing every
|
|
# second is 86 400 rows a day raw, and nobody reads a row of that.
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class MetricBucket(SQLModel, table=True):
|
|
"""One node's minute: how much ran, how long it took, how late it was."""
|
|
|
|
__tablename__ = "metric_minute"
|
|
|
|
flow: str = Field(primary_key=True, max_length=255)
|
|
node: str = Field(primary_key=True, max_length=255)
|
|
bucket: datetime = Field(
|
|
primary_key=True,
|
|
index=True,
|
|
sa_type=DateTime(timezone=True), # type: ignore
|
|
)
|
|
executions: int = 0
|
|
errors: int = 0
|
|
#: Messages emitted, which is not the same as executions: a node can run
|
|
#: and publish nothing.
|
|
messages: int = 0
|
|
duration_sum_ms: float = 0.0
|
|
duration_max_ms: float = 0.0
|
|
#: How long queued work waited before it ran.
|
|
lag_sum_ms: float = 0.0
|
|
lag_max_ms: float = 0.0
|
|
#: Items the lag sums are over, so an average can be taken.
|
|
items: int = 0
|
|
|
|
|
|
class EngineEvent(SQLModel, table=True):
|
|
"""Something worth keeping after the websocket has forgotten it."""
|
|
|
|
__tablename__ = "engine_event"
|
|
|
|
id: int | None = Field(default=None, primary_key=True)
|
|
ts: datetime = Field(
|
|
index=True,
|
|
sa_type=DateTime(timezone=True), # type: ignore
|
|
)
|
|
#: node_error, flow_quarantined, engine_degraded, …, or audit.
|
|
type: str = Field(max_length=32, index=True)
|
|
flow: str = ""
|
|
node: str = ""
|
|
#: The error and its traceback, or what an audit entry says was done.
|
|
detail: str = ""
|
|
#: Who did it, on audit rows.
|
|
actor: str = ""
|
|
|
|
|
|
class FlowRun(SQLModel, table=True):
|
|
"""One cascade, from the item that started it to the last node in it."""
|
|
|
|
__tablename__ = "flow_run"
|
|
|
|
#: The queue entry id, or a `manual-` one for a run that never queued.
|
|
id: str = Field(primary_key=True, max_length=64)
|
|
flow: str = Field(index=True, max_length=255)
|
|
#: What caused it: the work item's cause, or "manual".
|
|
source: str = ""
|
|
started_at: datetime = Field(
|
|
index=True,
|
|
sa_type=DateTime(timezone=True), # type: ignore
|
|
)
|
|
finished_at: datetime | None = Field(
|
|
default=None,
|
|
sa_type=DateTime(timezone=True), # type: ignore
|
|
)
|
|
#: running, ok, error or abandoned.
|
|
status: str = "running"
|
|
nodes: int = 0
|
|
errors: int = 0
|
|
duration_ms: float = 0.0
|
|
deliveries: int = 1
|