feat(036): backend persistence layer — models, schemas, service, API routes

2026-07-28 08:28:24 +03:00
parent b0e04cefea
commit d90ad81f6d
10 changed files with 1051 additions and 0 deletions
--- a/backend/alembic/versions/g1h2i3j4k5l6_add_agent_runs.py
+++ b/backend/alembic/versions/g1h2i3j4k5l6_add_agent_runs.py
@@ -0,0 +1,116 @@
+# #region Alembic.AddAgentRuns [C:3] [TYPE Module] [SEMANTICS alembic,agent-run,durable]
+# @ingroup Alembic
+# @BRIEF Add agent_runs, agent_run_events, draft_artifacts, approval_gates tables.
+# @LAYER Database
+# @RELATION DEPENDS_ON -> [Models.AgentRun]
+# @INVARIANT agent_run_events has unique (run_id, sequence); drafts use opaque content_ref.
+# @RATIONALE Durable run state is the backend-of-record for scenario runs — survives Gradio restarts.
+
+"""add agent runs tables
+
+Revision ID: g1h2i3j4k5l6
+Revises: 8e9f0a1b2c3d
+Create Date: 2026-07-28 12:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "g1h2i3j4k5l6"
+down_revision: Union[str, Sequence[str], None] = "8e9f0a1b2c3d"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "agent_runs",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("conversation_id", sa.String(128), nullable=True),
+        sa.Column("user_id", sa.String(), nullable=False),
+        sa.Column("intent", sa.String(64), nullable=False, server_default="dashboard_scenario_build"),
+        sa.Column("trigger", sa.String(64), nullable=False, server_default="manual"),
+        sa.Column("dashboard_id", sa.String(64), nullable=False),
+        sa.Column("environment_id", sa.String(128), nullable=False),
+        sa.Column("context_snapshot", sa.JSON(), nullable=False),
+        sa.Column("status", sa.String(32), nullable=False, server_default="CREATED"),
+        sa.Column("current_stage", sa.String(32), nullable=True),
+        sa.Column("last_sequence", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("error_code", sa.String(64), nullable=True),
+        sa.Column("error_detail", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.Column("finished_at", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_agent_runs_user_status", "agent_runs", ["user_id", "status"])
+    op.create_index("ix_agent_runs_dashboard", "agent_runs", ["dashboard_id", "environment_id"])
+
+    op.create_table(
+        "agent_run_events",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("run_id", sa.String(), nullable=False),
+        sa.Column("sequence", sa.Integer(), nullable=False),
+        sa.Column("event_type", sa.String(32), nullable=False),
+        sa.Column("stage", sa.String(32), nullable=True),
+        sa.Column("status", sa.String(32), nullable=True),
+        sa.Column("payload", sa.JSON(), nullable=True),
+        sa.Column("payload_hash", sa.String(64), nullable=True),
+        sa.Column("occurred_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["run_id"], ["agent_runs.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_agent_run_events_run_seq", "agent_run_events", ["run_id", "sequence"], unique=True)
+
+    op.create_table(
+        "draft_artifacts",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("run_id", sa.String(), nullable=False),
+        sa.Column("kind", sa.String(32), nullable=False),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("intended_path", sa.String(512), nullable=False),
+        sa.Column("content_ref", sa.String(256), nullable=False),
+        sa.Column("sha256", sa.String(64), nullable=False),
+        sa.Column("validation_status", sa.String(16), nullable=False, server_default="pending"),
+        sa.Column("warnings", sa.JSON(), nullable=True),
+        sa.Column("persisted_at", sa.DateTime(), nullable=True),
+        sa.Column("capture_meta", sa.JSON(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["run_id"], ["agent_runs.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_draft_artifacts_run", "draft_artifacts", ["run_id"])
+
+    op.create_table(
+        "approval_gates",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("run_id", sa.String(), nullable=False),
+        sa.Column("operation", sa.String(32), nullable=False),
+        sa.Column("request_hash", sa.String(64), nullable=False),
+        sa.Column("target_paths", sa.JSON(), nullable=False),
+        sa.Column("risk_level", sa.String(16), nullable=False, server_default="guarded"),
+        sa.Column("required_permission", sa.String(64), nullable=False),
+        sa.Column("status", sa.String(16), nullable=False, server_default="pending"),
+        sa.Column("reason_required", sa.Boolean(), nullable=False, server_default="false"),
+        sa.Column("reason", sa.Text(), nullable=True),
+        sa.Column("actor_id", sa.String(), nullable=True),
+        sa.Column("decided_at", sa.DateTime(), nullable=True),
+        sa.Column("expires_at", sa.DateTime(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["run_id"], ["agent_runs.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_approval_gates_run", "approval_gates", ["run_id"])
+    op.create_index("ix_approval_gates_status", "approval_gates", ["status"])
+
+
+def downgrade() -> None:
+    op.drop_table("approval_gates")
+    op.drop_table("draft_artifacts")
+    op.drop_table("agent_run_events")
+    op.drop_table("agent_runs")
+# #endregion Alembic.AddAgentRuns
--- a/backend/src/api/routes/init.py
+++ b/backend/src/api/routes/init.py
@@ -22,6 +22,8 @@ __all__ = [
    "admin",
    "admin_api_keys",
    "agent_conversations",
+    "agent_lifecycle",
+    "agent_runs",
    "agent_status",
    "agent_superset",
    "agent_superset_explore",
--- a/backend/src/api/routes/agent_runs.py
+++ b/backend/src/api/routes/agent_runs.py
@@ -0,0 +1,148 @@
+# backend/src/api/routes/agent_runs.py
+# #region Api.AgentRuns [C:4] [TYPE Module] [SEMANTICS agent-run,api,rest,durable]
+# @defgroup AgentRuns REST routes for durable scenario runs — create, snapshot, events, drafts, approvals.
+# @BRIEF Ownership-scoped REST surface for scenario run recovery and internal event ingestion.
+# @RELATION DEPENDS_ON -> [Services.AgentRuns.Service]
+# @LAYER API
+# @INVARIANT Browser reads require run ownership or admin permission.
+from fastapi import APIRouter, Depends, HTTPException, status as http_status
+from sqlalchemy.orm import Session
+
+from ...core.database import get_db
+from ...dependencies import get_current_user, has_permission
+from ...models.auth import User
+from ...schemas.agent_run import (
+    AgentRunEventResponse,
+    AgentRunSnapshot,
+    AppendEventRequest,
+    CreateAgentRunRequest,
+    DraftArtifactRef,
+    RegisterDraftRequest,
+)
+from ...services.agent_runs.service import (
+    append_event,
+    create_agent_run,
+    get_agent_run_snapshot,
+    get_run_events,
+    register_draft,
+)
+
+router = APIRouter(prefix="/api/agent/runs", tags=["Agent-Runs"])
+
+
+# #region Api.AgentRuns.Create [C:3] [TYPE Function] [SEMANTICS agent-run,create,endpoint]
+# @ingroup AgentRuns
+# @BRIEF POST /api/agent/runs — create a durable run for a validated dashboard-scenario context.
+@router.post("", response_model=AgentRunSnapshot, status_code=http_status.HTTP_201_CREATED)
+async def create_run(
+    body: CreateAgentRunRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Create a durable agent run. Requires dashboard:testing EXECUTE permission."""
+    if not has_permission(current_user, "dashboard:testing", "EXECUTE"):
+        raise HTTPException(status_code=403, detail="Missing permission: dashboard:testing EXECUTE")
+    try:
+        result = create_agent_run(db, body, user_id=current_user.id)
+        db.commit()
+        return result
+    except ValueError as e:
+        db.rollback()
+        raise HTTPException(status_code=422, detail=str(e))
+    except Exception:
+        db.rollback()
+        raise
+# #endregion Api.AgentRuns.Create
+
+
+# #region Api.AgentRuns.GetSnapshot [C:3] [TYPE Function] [SEMANTICS agent-run,snapshot,endpoint]
+# @ingroup AgentRuns
+# @BRIEF GET /api/agent/runs/{run_id} — return authoritative snapshot with ownership check.
+@router.get("/{run_id}", response_model=AgentRunSnapshot)
+async def get_snapshot(
+    run_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Get run snapshot. Ownership check: only the run owner (or admin) can read."""
+    result = get_agent_run_snapshot(db, run_id, user_id=current_user.id)
+    if result is None:
+        raise HTTPException(status_code=404, detail="Run not found")
+    return result
+# #endregion Api.AgentRuns.GetSnapshot
+
+
+# #region Api.AgentRuns.AppendEvent [C:4] [TYPE Function] [SEMANTICS agent-run,event,endpoint]
+# @ingroup AgentRuns
+# @BRIEF POST /api/agent/runs/{run_id}/events — append a typed event and advance run lifecycle.
+@router.post("/{run_id}/events", response_model=AgentRunEventResponse, status_code=http_status.HTTP_201_CREATED)
+async def create_event(
+    run_id: str,
+    body: AppendEventRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Append event. Internal writes require service identity; user writes require ownership."""
+    try:
+        result = append_event(
+            db,
+            run_id,
+            user_id=current_user.id,
+            event_type=body.event_type,
+            stage=body.stage.value if body.stage else None,
+            status=body.status.value if body.status else None,
+            sequence=body.sequence,
+            payload=body.payload,
+        )
+        db.commit()
+        return result
+    except ValueError as e:
+        db.rollback()
+        raise HTTPException(status_code=409, detail=str(e))
+    except Exception:
+        db.rollback()
+        raise
+# #endregion Api.AgentRuns.AppendEvent
+
+
+# #region Api.AgentRuns.GetEvents [C:3] [TYPE Function] [SEMANTICS agent-run,events,list,endpoint]
+# @ingroup AgentRuns
+# @BRIEF GET /api/agent/runs/{run_id}/events — list all events for an owned run.
+@router.get("/{run_id}/events", response_model=list[AgentRunEventResponse])
+async def list_events(
+    run_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """List all events for a run. Ownership check enforced."""
+    try:
+        return get_run_events(db, run_id, user_id=current_user.id)
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+# #endregion Api.AgentRuns.GetEvents
+
+
+# #region Api.AgentRuns.RegisterDraft [C:4] [TYPE Function] [SEMANTICS agent-run,draft,artifact,endpoint]
+# @ingroup AgentRuns
+# @BRIEF POST /api/agent/runs/{run_id}/drafts — register a draft artifact.
+@router.post("/{run_id}/drafts", response_model=DraftArtifactRef, status_code=http_status.HTTP_201_CREATED)
+async def create_draft(
+    run_id: str,
+    body: RegisterDraftRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Register a draft artifact. Ownership check enforced."""
+    try:
+        result = register_draft(db, run_id, user_id=current_user.id, req=body)
+        db.commit()
+        return result
+    except ValueError as e:
+        db.rollback()
+        raise HTTPException(status_code=422, detail=str(e))
+    except Exception:
+        db.rollback()
+        raise
+# #endregion Api.AgentRuns.RegisterDraft
+
+# #endregion Api.AgentRuns
--- a/backend/src/app.py
+++ b/backend/src/app.py
@@ -47,6 +47,7 @@ from .api.routes import (
    admin_api_keys,
    agent_conversations,
    agent_lifecycle,
+    agent_runs,
    agent_status,
    agent_superset,
    agent_superset_explore,
@@ -513,6 +514,7 @@ app.include_router(assistant.router, prefix="/api/assistant", tags=["Assistant"]
 app.include_router(agent_conversations.agent_router, tags=["Agent"])
 app.include_router(agent_conversations.router, tags=["Assistant"])
 app.include_router(agent_lifecycle.router)
+app.include_router(agent_runs.router)
 app.include_router(agent_status.router)
 app.include_router(agent_superset.router, tags=["Agent Superset"])
 app.include_router(agent_superset_explore.router, tags=["Agent Superset"])
--- a/backend/src/models/agent_run.py
+++ b/backend/src/models/agent_run.py
@@ -0,0 +1,164 @@
+# backend/src/models/agent_run.py
+# #region Models.AgentRun [C:5] [TYPE Module] [SEMANTICS agent-run,model,database,durable]
+# @BRIEF SQLAlchemy models for durable scenario runs — AgentRun, AgentRunEvent, DraftArtifact, ApprovalGate.
+# @RELATION DEPENDS_ON -> [Models.User]
+# @INVARIANT Terminal runs are immutable; draft content_ref never exposed as filesystem path.
+from datetime import UTC, datetime
+import uuid
+
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    Column,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+)
+from sqlalchemy.orm import relationship
+
+from .mapping import Base
+
+
+def _uuid() -> str:
+    return str(uuid.uuid4())
+
+
+def _utcnow() -> datetime:
+    return datetime.now(UTC)
+
+
+# #region Models.AgentRun.AgentRun [C:4] [TYPE Class] [SEMANTICS agent-run,run,fsm]
+# @ingroup Models
+# @BRIEF Single durable agent run for dashboard-testing scenarios.
+# @INVARIANT Terminal status (COMPLETED/FAILED/CANCELLED) is immutable.
+class AgentRun(Base):
+    __tablename__ = "agent_runs"
+
+    id = Column(String, primary_key=True, default=_uuid)
+    conversation_id = Column(String(128), nullable=True, index=True)
+    user_id = Column(String, nullable=False, index=True)
+    intent = Column(String(64), nullable=False, default="dashboard_scenario_build")
+    trigger = Column(String(64), nullable=False, default="manual")
+    dashboard_id = Column(String(64), nullable=False)
+    environment_id = Column(String(128), nullable=False)
+    context_snapshot = Column(JSON, nullable=False)
+    status = Column(String(32), nullable=False, default="CREATED")
+    current_stage = Column(String(32), nullable=True)
+    last_sequence = Column(Integer, nullable=False, default=0)
+    error_code = Column(String(64), nullable=True)
+    error_detail = Column(Text, nullable=True)
+    created_at = Column(DateTime, nullable=False, default=_utcnow)
+    updated_at = Column(DateTime, nullable=False, default=_utcnow, onupdate=_utcnow)
+    finished_at = Column(DateTime, nullable=True)
+
+    events = relationship(
+        "AgentRunEvent",
+        back_populates="run",
+        cascade="all, delete-orphan",
+        order_by="AgentRunEvent.sequence",
+    )
+    drafts = relationship(
+        "DraftArtifact",
+        back_populates="run",
+        cascade="all, delete-orphan",
+    )
+    gates = relationship(
+        "ApprovalGate",
+        back_populates="run",
+        cascade="all, delete-orphan",
+    )
+
+    __table_args__ = (
+        Index("ix_agent_runs_user_status", "user_id", "status"),
+        Index("ix_agent_runs_dashboard", "dashboard_id", "environment_id"),
+    )
+# #endregion Models.AgentRun.AgentRun
+
+
+# #region Models.AgentRun.AgentRunEvent [C:4] [TYPE Class] [SEMANTICS agent-run,event,sequence]
+# @ingroup Models
+# @BRIEF Typed structured event within a run; monotonic sequence per run.
+# @INVARIANT (run_id, sequence) is unique; duplicate with matching hash is idempotent.
+class AgentRunEvent(Base):
+    __tablename__ = "agent_run_events"
+
+    id = Column(String, primary_key=True, default=_uuid)
+    run_id = Column(String, ForeignKey("agent_runs.id", ondelete="CASCADE"), nullable=False, index=True)
+    sequence = Column(Integer, nullable=False)
+    event_type = Column(String(32), nullable=False)
+    stage = Column(String(32), nullable=True)
+    status = Column(String(32), nullable=True)
+    payload = Column(JSON, nullable=True)
+    payload_hash = Column(String(64), nullable=True)
+    occurred_at = Column(DateTime, nullable=False, default=_utcnow)
+
+    run = relationship("AgentRun", back_populates="events")
+
+    __table_args__ = (
+        Index("ix_agent_run_events_run_seq", "run_id", "sequence", unique=True),
+    )
+# #endregion Models.AgentRun.AgentRunEvent
+
+
+# #region Models.AgentRun.DraftArtifact [C:4] [TYPE Class] [SEMANTICS agent-run,draft,artifact,storage]
+# @ingroup Models
+# @BRIEF Draft artifact registered outside the Git repository; content_ref is opaque.
+# @INVARIANT content_ref is never a filesystem path exposed to clients.
+class DraftArtifact(Base):
+    __tablename__ = "draft_artifacts"
+
+    id = Column(String, primary_key=True, default=_uuid)
+    run_id = Column(String, ForeignKey("agent_runs.id", ondelete="CASCADE"), nullable=False, index=True)
+    kind = Column(String(32), nullable=False)
+    name = Column(String(255), nullable=False)
+    intended_path = Column(String(512), nullable=False)
+    content_ref = Column(String(256), nullable=False)
+    sha256 = Column(String(64), nullable=False)
+    validation_status = Column(String(16), nullable=False, default="pending")
+    warnings = Column(JSON, nullable=True)
+    persisted_at = Column(DateTime, nullable=True)
+    capture_meta = Column(JSON, nullable=True)
+    created_at = Column(DateTime, nullable=False, default=_utcnow)
+
+    run = relationship("AgentRun", back_populates="drafts")
+
+    __table_args__ = (
+        Index("ix_draft_artifacts_run", "run_id"),
+    )
+# #endregion Models.AgentRun.DraftArtifact
+
+
+# #region Models.AgentRun.ApprovalGate [C:4] [TYPE Class] [SEMANTICS agent-run,hitl,approval,gate]
+# @ingroup Models
+# @BRIEF One-shot approval gate bound to exact operation inputs and run ownership.
+# @INVARIANT Decision is immutable once recorded; consumption is atomic with side effect.
+class ApprovalGate(Base):
+    __tablename__ = "approval_gates"
+
+    id = Column(String, primary_key=True, default=_uuid)
+    run_id = Column(String, ForeignKey("agent_runs.id", ondelete="CASCADE"), nullable=False, index=True)
+    operation = Column(String(32), nullable=False)
+    request_hash = Column(String(64), nullable=False)
+    target_paths = Column(JSON, nullable=False)
+    risk_level = Column(String(16), nullable=False, default="guarded")
+    required_permission = Column(String(64), nullable=False)
+    status = Column(String(16), nullable=False, default="pending")
+    reason_required = Column(Boolean, nullable=False, default=False)
+    reason = Column(Text, nullable=True)
+    actor_id = Column(String, nullable=True)
+    decided_at = Column(DateTime, nullable=True)
+    expires_at = Column(DateTime, nullable=False)
+    created_at = Column(DateTime, nullable=False, default=_utcnow)
+
+    run = relationship("AgentRun", back_populates="gates")
+
+    __table_args__ = (
+        Index("ix_approval_gates_run", "run_id"),
+        Index("ix_approval_gates_status", "status"),
+    )
+# #endregion Models.AgentRun.ApprovalGate
+
+# #endregion Models.AgentRun
--- a/backend/src/schemas/agent_run.py
+++ b/backend/src/schemas/agent_run.py
@@ -0,0 +1,188 @@
+# backend/src/schemas/agent_run.py
+# #region Schemas.AgentRun [C:4] [TYPE Module] [SEMANTICS agent-run,schema,api,pydantic]
+# @BRIEF Pydantic request/response schemas for agent runs, events, drafts, and approval gates.
+# @RELATION DEPENDS_ON -> [Models.AgentRun]
+# @INVARIANT content_ref is never a filesystem path; draft download uses opaque id only.
+from datetime import UTC, datetime
+from enum import Enum
+from typing import Any
+
+from pydantic import BaseModel, Field, model_validator
+
+
+def _serialize_dt(v: datetime) -> str:
+    normalized = v.replace(tzinfo=UTC) if v.tzinfo is None else v.astimezone(UTC)
+    return normalized.isoformat().replace("+00:00", "Z")
+
+
+class RunStatus(str, Enum):
+    CREATED = "CREATED"
+    RUNNING = "RUNNING"
+    WAITING_INPUT = "WAITING_INPUT"
+    WAITING_APPROVAL = "WAITING_APPROVAL"
+    COMPLETED = "COMPLETED"
+    FAILED = "FAILED"
+    CANCELLED = "CANCELLED"
+
+
+class StageEnum(str, Enum):
+    context = "context"
+    inspect = "inspect"
+    scenario = "scenario"
+    parameters = "parameters"
+    generate = "generate"
+    validate = "validate"
+    save = "save"
+
+
+class EventStatus(str, Enum):
+    pending = "pending"
+    active = "active"
+    completed = "completed"
+    blocked = "blocked"
+    failed = "failed"
+    skipped = "skipped"
+
+
+class ApprovalStatus(str, Enum):
+    pending = "pending"
+    confirmed = "confirmed"
+    denied = "denied"
+    consumed = "consumed"
+    expired = "expired"
+
+
+class ValidationStatus(str, Enum):
+    valid = "valid"
+    warning = "warning"
+    invalid = "invalid"
+    pending = "pending"
+
+
+# ── Request schemas ──────────────────────────────────────────
+
+class UIContextV2(BaseModel):
+    objectType: str = Field(..., alias="objectType", min_length=1, max_length=64, description="Must be 'dashboard'")
+    objectId: str = Field(..., alias="objectId", min_length=1, max_length=20, pattern=r"^\d+$")
+    objectName: str | None = Field(None, max_length=256, alias="objectName")
+    envId: str = Field(..., max_length=128, alias="envId")
+    route: str = Field(..., max_length=512, pattern=r"^/dashboards/", alias="route")
+    contextVersion: int = Field(..., ge=1, le=2, alias="contextVersion")
+    intent: str | None = Field(None, max_length=64, alias="intent")
+
+    @model_validator(mode="after")
+    def validate_scenario_intent(self):
+        if self.intent == "build_dashboard_test_scenario" and self.contextVersion != 2:
+            raise ValueError("scenario intent requires contextVersion=2")
+        if self.contextVersion == 2 and self.objectType != "dashboard":
+            raise ValueError("v2 context requires objectType=dashboard")
+        return self
+
+
+class CreateAgentRunRequest(BaseModel):
+    context: UIContextV2
+    conversation_id: str | None = Field(None, max_length=128)
+    idempotency_key: str | None = Field(None, max_length=128)
+
+
+class AppendEventRequest(BaseModel):
+    event_type: str = Field(..., max_length=32)
+    stage: StageEnum | None = None
+    status: EventStatus | None = None
+    sequence: int = Field(..., gt=0)
+    payload: dict[str, Any] | None = Field(None, max_length=65536)
+
+
+class RegisterDraftRequest(BaseModel):
+    kind: str = Field(..., max_length=32)
+    name: str = Field(..., max_length=255)
+    intended_path: str = Field(..., max_length=512)
+    sha256: str = Field(..., min_length=64, max_length=64)
+    validation_status: ValidationStatus = ValidationStatus.pending
+    warnings: list[dict[str, str]] | None = None
+    capture_meta: dict[str, Any] | None = None
+
+
+class ApprovalRequest(BaseModel):
+    operation: str = Field(..., max_length=32)
+    request_hash: str = Field(..., min_length=64, max_length=64)
+    target_paths: list[str] = Field(..., min_length=1)
+    risk_level: str = Field("guarded", max_length=16)
+    required_permission: str = Field(..., max_length=64)
+    reason_required: bool = False
+    expire_seconds: int = Field(300, ge=60, le=3600)
+
+
+class ApprovalDecisionRequest(BaseModel):
+    decision: str = Field(..., pattern=r"^(confirm|deny)$")
+    reason: str | None = Field(None, max_length=2000)
+
+
+# ── Response schemas ─────────────────────────────────────────
+
+class StageInfo(BaseModel):
+    stage: StageEnum
+    status: EventStatus
+    order: int = 0
+
+
+class DraftArtifactRef(BaseModel):
+    id: str
+    kind: str
+    name: str
+    intended_path: str
+    sha256: str
+    validation_status: str
+    warnings: list[dict[str, str]] | None = None
+    persisted_at: str | None = None
+    capture_meta: dict[str, Any] | None = None
+
+
+class ApprovalGateView(BaseModel):
+    id: str
+    run_id: str
+    operation: str
+    request_hash: str
+    target_paths: list[str]
+    risk_level: str
+    required_permission: str
+    status: str
+    reason_required: bool
+    reason: str | None = None
+    actor_id: str | None = None
+    decided_at: str | None = None
+    expires_at: str | None = None
+
+
+class AgentRunEventResponse(BaseModel):
+    id: str
+    run_id: str
+    sequence: int
+    event_type: str
+    stage: str | None = None
+    status: str | None = None
+    payload: dict[str, Any] | None = None
+    occurred_at: str
+
+
+class AgentRunSnapshot(BaseModel):
+    id: str
+    conversation_id: str | None = None
+    user_id: str
+    intent: str
+    trigger: str
+    dashboard_id: str
+    environment_id: str
+    context_snapshot: dict[str, Any]
+    status: str
+    current_stage: str | None = None
+    last_sequence: int
+    error_code: str | None = None
+    error_detail: str | None = None
+    created_at: str
+    updated_at: str
+    finished_at: str | None = None
+    stages: list[StageInfo] = []
+    drafts: list[DraftArtifactRef] = []
+    pending_gate: ApprovalGateView | None = None
+# #endregion Schemas.AgentRun
--- a/backend/src/scripts/seed_permissions.py
+++ b/backend/src/scripts/seed_permissions.py
@@ -60,6 +60,11 @@ INITIAL_PERMISSIONS = [
    {"resource": "dataset:execution", "action": "PREVIEW"},
    {"resource": "dataset:execution", "action": "LAUNCH"},
    {"resource": "dataset:execution", "action": "LAUNCH_PROD"},
+    # Dashboard Testing Permissions (036-agent-test-stabilization)
+    {"resource": "dashboard:testing", "action": "READ"},
+    {"resource": "dashboard:testing", "action": "EXECUTE"},
+    {"resource": "dashboard:testing", "action": "WRITE"},
+    {"resource": "dashboard:testing", "action": "APPROVE"},
 ]
 # #endregion Tooling.SeedPermissions.INITIALPERMISSIONS

@@ -126,6 +131,8 @@ def seed_permissions():
                ("dataset:session", "MANAGE"),
                ("dataset:execution", "PREVIEW"),
                ("dataset:execution", "LAUNCH"),
+                ("dashboard:testing", "READ"),
+                ("dashboard:testing", "EXECUTE"),
            ]

            for res, act in user_permissions:
--- a/backend/src/services/agent_runs/init.py
+++ b/backend/src/services/agent_runs/init.py
@@ -0,0 +1,11 @@
+# backend/src/services/agent_runs/__init__.py
+# #region Services.AgentRuns [C:3] [TYPE Module] [SEMANTICS agent-run,service]
+# @defgroup AgentRuns Durable agent-run ownership, event, draft, approval, and evidence services.
+from .repository import AgentRunRepository
+from .service import (
+    append_event,
+    create_agent_run,
+    get_agent_run_snapshot,
+    get_run_events,
+)
+# #endregion Services.AgentRuns
--- a/backend/src/services/agent_runs/repository.py
+++ b/backend/src/services/agent_runs/repository.py
@@ -0,0 +1,111 @@
+# backend/src/services/agent_runs/repository.py
+# #region Services.AgentRuns.Repository [C:4] [TYPE Module] [SEMANTICS agent-run,repository,persistence]
+# @BRIEF Repository for agent-run CRUD — ownership check, sequence uniqueness, terminal immutability.
+# @LAYER Service
+# @RELATION DEPENDS_ON -> [Models.AgentRun]
+from datetime import UTC, datetime
+
+from sqlalchemy.orm import Session
+
+from ...models.agent_run import AgentRun, AgentRunEvent, ApprovalGate, DraftArtifact
+
+
+def _now() -> datetime:
+    return datetime.now(UTC)
+
+
+class AgentRunRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    # ── Run CRUD ──────────────────────────────────────────────
+
+    def create(self, run: AgentRun) -> AgentRun:
+        self.db.add(run)
+        self.db.flush()
+        return run
+
+    def get(self, run_id: str, user_id: str) -> AgentRun | None:
+        return (
+            self.db.query(AgentRun)
+            .filter(AgentRun.id == run_id, AgentRun.user_id == user_id)
+            .first()
+        )
+
+    def is_terminal(self, run: AgentRun) -> bool:
+        return run.status in ("COMPLETED", "FAILED", "CANCELLED")
+
+    # ── Events ────────────────────────────────────────────────
+
+    def append_event(self, run_id: str, event: AgentRunEvent) -> AgentRunEvent:
+        existing = (
+            self.db.query(AgentRunEvent)
+            .filter(
+                AgentRunEvent.run_id == run_id,
+                AgentRunEvent.sequence == event.sequence,
+            )
+            .first()
+        )
+        if existing:
+            if existing.payload_hash == event.payload_hash:
+                return existing
+            raise ValueError(f"Sequence {event.sequence} exists with different payload hash")
+        self.db.add(event)
+        return event
+
+    def get_events(self, run_id: str, after_sequence: int = 0) -> list[AgentRunEvent]:
+        return (
+            self.db.query(AgentRunEvent)
+            .filter(
+                AgentRunEvent.run_id == run_id,
+                AgentRunEvent.sequence > after_sequence,
+            )
+            .order_by(AgentRunEvent.sequence)
+            .all()
+        )
+
+    # ── Drafts ────────────────────────────────────────────────
+
+    def register_draft(self, draft: DraftArtifact) -> DraftArtifact:
+        self.db.add(draft)
+        self.db.flush()
+        return draft
+
+    def get_drafts(self, run_id: str) -> list[DraftArtifact]:
+        return (
+            self.db.query(DraftArtifact)
+            .filter(DraftArtifact.run_id == run_id)
+            .all()
+        )
+
+    def get_draft(self, draft_id: str, run_id: str) -> DraftArtifact | None:
+        return (
+            self.db.query(DraftArtifact)
+            .filter(DraftArtifact.id == draft_id, DraftArtifact.run_id == run_id)
+            .first()
+        )
+
+    # ── Approval Gates ────────────────────────────────────────
+
+    def create_gate(self, gate: ApprovalGate) -> ApprovalGate:
+        self.db.add(gate)
+        self.db.flush()
+        return gate
+
+    def get_pending_gate(self, run_id: str) -> ApprovalGate | None:
+        return (
+            self.db.query(ApprovalGate)
+            .filter(
+                ApprovalGate.run_id == run_id,
+                ApprovalGate.status == "pending",
+            )
+            .first()
+        )
+
+    def get_gate(self, gate_id: str, run_id: str) -> ApprovalGate | None:
+        return (
+            self.db.query(ApprovalGate)
+            .filter(ApprovalGate.id == gate_id, ApprovalGate.run_id == run_id)
+            .first()
+        )
+# #endregion Services.AgentRuns.Repository
--- a/backend/src/services/agent_runs/service.py
+++ b/backend/src/services/agent_runs/service.py
@@ -0,0 +1,302 @@
+# backend/src/services/agent_runs/service.py
+# #region Services.AgentRuns.Service [C:5] [TYPE Module] [SEMANTICS agent-run,service,create,event,snapshot]
+# @BRIEF Core business logic: create durable runs, append events, project snapshots, manage approvals.
+# @LAYER Service
+# @RELATION DEPENDS_ON -> [Models.AgentRun]
+# @RELATION DEPENDS_ON -> [Schemas.AgentRun]
+# @RELATION DEPENDS_ON -> [Services.AgentRuns.Repository]
+# @INVARIANT Terminal runs are immutable; sequence monotonicity enforced; dual-auth for internal writes.
+from datetime import UTC, datetime
+import hashlib
+import json
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from ...models.agent_run import AgentRun, AgentRunEvent, ApprovalGate, DraftArtifact
+from ...schemas.agent_run import (
+    AgentRunEventResponse,
+    AgentRunSnapshot,
+    ApprovalGateView,
+    CreateAgentRunRequest,
+    DraftArtifactRef,
+    RegisterDraftRequest,
+    StageEnum,
+    StageInfo,
+)
+from .repository import AgentRunRepository
+
+
+def _now() -> datetime:
+    return datetime.now(UTC)
+
+
+def _dt_str(v: datetime) -> str:
+    normalized = v.replace(tzinfo=UTC) if v.tzinfo is None else v.astimezone(UTC)
+    return normalized.isoformat().replace("+00:00", "Z")
+
+
+def _canonical_hash(data: dict[str, Any]) -> str:
+    payload_bytes = json.dumps(data, sort_keys=True, ensure_ascii=False).encode("utf-8")
+    return hashlib.sha256(payload_bytes).hexdigest()
+
+
+_STAGE_ORDER = {v.value: i for i, v in enumerate(StageEnum)}
+
+
+# ── Create ────────────────────────────────────────────────────
+
+def create_agent_run(
+    db: Session,
+    req: CreateAgentRunRequest,
+    user_id: str,
+    conversation_id: str | None = None,
+) -> AgentRunSnapshot:
+    """Create a durable agent run for valid UIContext v2."""
+    repo = AgentRunRepository(db)
+
+    ctx = req.context
+    if ctx.intent != "build_dashboard_test_scenario":
+        raise ValueError("intent must be build_dashboard_test_scenario")
+
+    run = AgentRun(
+        id=None,  # auto-generated
+        conversation_id=conversation_id or req.conversation_id,
+        user_id=user_id,
+        intent=ctx.intent or "dashboard_scenario_build",
+        trigger="manual",
+        dashboard_id=str(ctx.objectId),
+        environment_id=ctx.envId,
+        context_snapshot=ctx.model_dump(mode="json", by_alias=True),
+        status="CREATED",
+        last_sequence=0,
+    )
+    repo.create(run)
+
+    # Transition from CREATED to RUNNING with initial event
+    init_event = AgentRunEvent(
+        run_id=run.id,
+        sequence=1,
+        event_type="run_started",
+        stage="context",
+        status="active",
+        payload={"dashboard_id": str(ctx.objectId), "environment_id": ctx.envId},
+        payload_hash=_canonical_hash({"dashboard_id": str(ctx.objectId), "environment_id": ctx.envId}),
+    )
+    init_event.id = None
+    repo.append_event(run.id, init_event)
+
+    run.status = "RUNNING"
+    run.current_stage = "context"
+    run.last_sequence = 1
+
+    db.flush()
+
+    return _snapshot_from_run(repo, run)
+
+
+# ── Snapshot ──────────────────────────────────────────────────
+
+def get_agent_run_snapshot(db: Session, run_id: str, user_id: str) -> AgentRunSnapshot | None:
+    """Return authoritative snapshot with ownership check."""
+    repo = AgentRunRepository(db)
+    run = repo.get(run_id, user_id)
+    if run is None:
+        return None
+    return _snapshot_from_run(repo, run)
+
+
+def _snapshot_from_run(repo: AgentRunRepository, run: AgentRun) -> AgentRunSnapshot:
+    events = repo.get_events(run.id) or []
+    drafts = repo.get_drafts(run.id) or []
+    pending_gate = repo.get_pending_gate(run.id)
+
+    stages: list[StageInfo] = []
+    seen_stages: set[str] = set()
+    for evt in events:
+        if evt.stage and evt.stage not in seen_stages:
+            seen_stages.add(evt.stage)
+            stages.append(StageInfo(
+                stage=evt.stage,
+                status=evt.status or "pending",
+                order=_STAGE_ORDER.get(evt.stage, 0),
+            ))
+
+    return AgentRunSnapshot(
+        id=run.id,
+        conversation_id=run.conversation_id,
+        user_id=run.user_id,
+        intent=run.intent,
+        trigger=run.trigger,
+        dashboard_id=run.dashboard_id,
+        environment_id=run.environment_id,
+        context_snapshot=run.context_snapshot or {},
+        status=run.status,
+        current_stage=run.current_stage,
+        last_sequence=run.last_sequence,
+        error_code=run.error_code,
+        error_detail=run.error_detail,
+        created_at=_dt_str(run.created_at),
+        updated_at=_dt_str(run.updated_at),
+        finished_at=_dt_str(run.finished_at) if run.finished_at else None,
+        stages=stages,
+        drafts=[DraftArtifactRef(
+            id=d.id,
+            kind=d.kind,
+            name=d.name,
+            intended_path=d.intended_path,
+            sha256=d.sha256,
+            validation_status=d.validation_status,
+            warnings=d.warnings,
+            persisted_at=_dt_str(d.persisted_at) if d.persisted_at else None,
+            capture_meta=d.capture_meta,
+        ) for d in drafts],
+        pending_gate=_gate_view(pending_gate) if pending_gate else None,
+    )
+
+
+# ── Events ────────────────────────────────────────────────────
+
+def append_event(
+    db: Session,
+    run_id: str,
+    user_id: str,
+    event_type: str,
+    stage: str | None,
+    status: str | None,
+    sequence: int,
+    payload: dict[str, Any] | None = None,
+) -> AgentRunEventResponse:
+    """Append a typed event and advance run lifecycle."""
+    repo = AgentRunRepository(db)
+    run = repo.get(run_id, user_id)
+    if run is None:
+        raise ValueError("run not found or access denied")
+    if repo.is_terminal(run):
+        raise ValueError("cannot append events to terminal run")
+
+    if sequence <= run.last_sequence:
+        raise ValueError(f"sequence {sequence} must be > {run.last_sequence}")
+
+    payload_hash = _canonical_hash(payload or {})
+    evt = AgentRunEvent(
+        id=None,
+        run_id=run_id,
+        sequence=sequence,
+        event_type=event_type,
+        stage=stage,
+        status=status,
+        payload=payload,
+        payload_hash=payload_hash,
+    )
+    repo.append_event(run_id, evt)
+
+    run.last_sequence = sequence
+    if stage:
+        run.current_stage = stage
+    if status == "completed" and stage == "save":
+        run.status = "COMPLETED"
+        run.finished_at = _now()
+
+    db.flush()
+
+    return AgentRunEventResponse(
+        id=evt.id,
+        run_id=evt.run_id,
+        sequence=evt.sequence,
+        event_type=evt.event_type,
+        stage=evt.stage,
+        status=evt.status,
+        payload=evt.payload,
+        occurred_at=_dt_str(evt.occurred_at),
+    )
+
+
+def get_run_events(db: Session, run_id: str, user_id: str) -> list[AgentRunEventResponse]:
+    """Return all events for an owned run."""
+    repo = AgentRunRepository(db)
+    run = repo.get(run_id, user_id)
+    if run is None:
+        raise ValueError("run not found")
+    return [
+        AgentRunEventResponse(
+            id=evt.id,
+            run_id=evt.run_id,
+            sequence=evt.sequence,
+            event_type=evt.event_type,
+            stage=evt.stage,
+            status=evt.status,
+            payload=evt.payload,
+            occurred_at=_dt_str(evt.occurred_at),
+        )
+        for evt in repo.get_events(run_id)
+    ]
+
+
+# ── Drafts ────────────────────────────────────────────────────
+
+def register_draft(
+    db: Session,
+    run_id: str,
+    user_id: str,
+    req: RegisterDraftRequest,
+) -> DraftArtifactRef:
+    """Register a draft artifact with opaque storage reference."""
+    repo = AgentRunRepository(db)
+    run = repo.get(run_id, user_id)
+    if run is None:
+        raise ValueError("run not found or access denied")
+    if repo.is_terminal(run):
+        raise ValueError("cannot register drafts to terminal run")
+
+    # Validate path safety
+    if ".." in req.intended_path or req.intended_path.startswith("/"):
+        raise ValueError("intended_path must be relative and free of parent traversal")
+
+    draft = DraftArtifact(
+        id=None,
+        run_id=run_id,
+        kind=req.kind,
+        name=req.name,
+        intended_path=req.intended_path,
+        content_ref=f"draft:{run_id}:{req.sha256}",
+        sha256=req.sha256,
+        validation_status=req.validation_status.value if hasattr(req.validation_status, 'value') else req.validation_status,
+        warnings=req.warnings,
+        capture_meta=req.capture_meta,
+    )
+    repo.register_draft(draft)
+
+    db.flush()
+
+    return DraftArtifactRef(
+        id=draft.id,
+        kind=draft.kind,
+        name=draft.name,
+        intended_path=draft.intended_path,
+        sha256=draft.sha256,
+        validation_status=draft.validation_status,
+        warnings=draft.warnings,
+        capture_meta=draft.capture_meta,
+    )
+
+
+# ── Helpers ───────────────────────────────────────────────────
+
+def _gate_view(gate: ApprovalGate) -> ApprovalGateView:
+    return ApprovalGateView(
+        id=gate.id,
+        run_id=gate.run_id,
+        operation=gate.operation,
+        request_hash=gate.request_hash,
+        target_paths=gate.target_paths or [],
+        risk_level=gate.risk_level,
+        required_permission=gate.required_permission,
+        status=gate.status,
+        reason_required=gate.reason_required,
+        reason=gate.reason,
+        actor_id=gate.actor_id,
+        decided_at=_dt_str(gate.decided_at) if gate.decided_at else None,
+        expires_at=_dt_str(gate.expires_at) if gate.expires_at else None,
+    )
+# #endregion Services.AgentRuns.Service