activity-core/src/activity_core/orm.py

183 lines
7.2 KiB
Python
Raw Normal View History

"""SQLAlchemy ORM table definitions for activity-core.
These are the persistence-layer counterparts to the Pydantic domain models in
models.py. Alembic reads Base.metadata (imported via db.py) for autogenerate.
"""
from __future__ import annotations
import uuid
from datetime import datetime
from sqlalchemy import (
Boolean,
DateTime,
ForeignKey,
Integer,
String,
Text,
func,
)
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.orm import Mapped, mapped_column
from activity_core.db import Base
class ActivityDefinition(Base):
__tablename__ = "activity_definitions"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
name: Mapped[str] = mapped_column(Text, nullable=False)
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
trigger_type: Mapped[str] = mapped_column(Text, nullable=False)
trigger_config: Mapped[dict] = mapped_column(JSONB, nullable=False)
context_sources: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
task_templates: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
feat(WP-0003b): parser, workflow wiring, triggers, webhooks T44: ActivityDefinition markdown file parser (definition_parser.py) - Scans activity-definitions/*.md and ACTIVITY_DEFINITION_DIRS paths - Parses YAML frontmatter + fenced rule/instruction blocks - Raises ParseError on any malformed file — never silently skips T45: ActivityDefinition sync command - Migration 0006: adds rules_json/instructions_json JSONB columns - sync_activity_definitions.py + make sync-activity-definitions - Called at worker startup before schedule sync T46: Rule/instruction pipeline wired into RunActivityWorkflow - New evaluate_rules and emit_tasks Temporal activities - Workflow passes event_envelope_json to enable rule evaluation - EventRouter now passes full envelope JSON as 4th workflow arg - IssueSink.emit() writes task_spawn_log rows per task T47: ScheduledTriggerConfig model (one-off future datetime trigger) T48: One-off Temporal Schedule support - Fixed timezone_name → time_zone_name (was causing all schedule tests to fail) - Added ScheduleCalendarSpec-based one-off schedule with remaining_actions=1 - cancel_scheduled() for admin cancellation - Fixed backfill() call to use *args unpacking (not list wrapper) - Fixed ScheduleAlreadyRunningError catch in upsert_schedule - sync_schedules now handles ScheduledTriggerConfig definitions T49: Webhook receiver - POST /webhooks/gitea — HMAC-SHA256 via X-Gitea-Signature-256 - POST /webhooks/github — HMAC-SHA256 via X-Hub-Signature-256 - Normalisers: repo.created, push, issue.closed → EventEnvelope - Publishes to NATS activity.{type} subject after registry validation - Mounted in api.py at /webhooks prefix T50: Gitea event type definitions - gitea.repo.created.md, gitea.push.md, gitea.issue.closed.md - Each includes normaliser field mapping in Consumer Notes Tests: 18 passed, 1 skipped (integration). Fixed embedded Temporal server visibility latency in test_upsert_schedule_creates_schedule. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-14 23:02:33 +02:00
rules_json: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
instructions_json: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
dedupe_key_strategy: Mapped[str] = mapped_column(
Text, nullable=False, default="skip"
)
version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
server_default=func.now(),
onupdate=func.now(),
)
class ActivityRun(Base):
__tablename__ = "activity_runs"
run_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
activity_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("activity_definitions.id", ondelete="RESTRICT"),
nullable=False,
index=True,
)
scheduled_for: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
fired_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
context_snapshot: Mapped[dict] = mapped_column(
JSONB, nullable=False, default=dict
)
tasks_spawned: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
version_used: Mapped[int] = mapped_column(Integer, nullable=False)
class TaskSpawnLog(Base):
__tablename__ = "task_spawn_log"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
activity_def_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("activity_definitions.id"),
nullable=False,
index=True,
)
source_type: Mapped[str] = mapped_column(String(20), nullable=False)
source_id: Mapped[str] = mapped_column(Text, nullable=False)
source_version: Mapped[str] = mapped_column(Text, nullable=False)
triggering_event_id: Mapped[str] = mapped_column(Text, nullable=False, index=True)
task_ref: Mapped[str | None] = mapped_column(Text, nullable=True)
condition_matched: Mapped[str | None] = mapped_column(Text, nullable=True)
prompt_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
model: Mapped[str | None] = mapped_column(Text, nullable=True)
output_validated: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
review_required: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
class EventType(Base):
__tablename__ = "event_types"
type_id: Mapped[str] = mapped_column(Text, primary_key=True)
version: Mapped[str] = mapped_column(Text, nullable=False)
publisher: Mapped[str] = mapped_column(Text, nullable=False)
governance: Mapped[str] = mapped_column(Text, nullable=False, default="publisher-declared")
status: Mapped[str] = mapped_column(Text, nullable=False, default="active")
attribute_schema: Mapped[dict] = mapped_column(JSONB, nullable=False)
raw_md: Mapped[str] = mapped_column(Text, nullable=False)
synced_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
class TaskInstance(Base):
__tablename__ = "task_instances"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
run_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("activity_runs.run_id", ondelete="CASCADE"),
nullable=False,
index=True,
)
type: Mapped[str] = mapped_column(Text, nullable=False)
params: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
status: Mapped[str] = mapped_column(Text, nullable=False, default="pending")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
class OpsRun(Base):
"""Claimable automation run instance (ACT-ADR-005 / ACTIVITY-WP-0026)."""
__tablename__ = "ops_runs"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
activity_definition_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("activity_definitions.id", ondelete="RESTRICT"),
nullable=False,
index=True,
)
idempotency_key: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
target_repo: Mapped[str | None] = mapped_column(Text, nullable=True)
title: Mapped[str] = mapped_column(Text, nullable=False)
description: Mapped[str] = mapped_column(Text, nullable=False, default="")
labels: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
priority: Mapped[str] = mapped_column(Text, nullable=False, default="medium")
state: Mapped[str] = mapped_column(Text, nullable=False, default="open", index=True)
claim_owner: Mapped[str | None] = mapped_column(Text, nullable=True)
lease_until: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, index=True
)
attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
source_type: Mapped[str] = mapped_column(Text, nullable=False)
source_id: Mapped[str] = mapped_column(Text, nullable=False)
triggering_event_id: Mapped[str] = mapped_column(Text, nullable=False, index=True)
approach_hint: Mapped[str | None] = mapped_column(Text, nullable=True)
result: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
server_default=func.now(),
onupdate=func.now(),
)