feat: schedule type system for work/entertainment periods

- New model: ScheduleType (name, work_from/to, entertainment_from/to)
- Agent.schedule_type_id FK to schedule_types
- CRUD API: GET/POST/PATCH/DELETE /schedule-types/
- Agent assignment: PUT /schedule-types/agent/{agent_id}/assign
- Agent self-query: GET /schedule-types/agent/me
- Permissions: schedule_type.read, schedule_type.manage
- Migration: adds schedule_type_id column to agents table

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

app/api/routers/schedule_type.py

@@ -0,0 +1,209 @@
+"""ScheduleType API router.
+
+CRUD for schedule types (work/entertainment time periods)
+and agent schedule type assignment.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException, Header
+from sqlalchemy.orm import Session
+from typing import List
+
+from app.core.config import get_db
+from app.api.deps import get_current_user
+from app.models.models import User
+from app.models.agent import Agent
+from app.models.schedule_type import ScheduleType
+from app.models.role_permission import Permission, RolePermission
+from app.schemas.schedule_type import (
+    ScheduleTypeCreate,
+    ScheduleTypeUpdate,
+    ScheduleTypeResponse,
+    AgentScheduleTypeAssign,
+)
+
+router = APIRouter(prefix="/schedule-types", tags=["ScheduleTypes"])
+
+
+# ---------------------------------------------------------------------------
+# Permission helpers
+# ---------------------------------------------------------------------------
+
+def _has_permission(db: Session, user: User, permission_name: str) -> bool:
+    if user.is_admin:
+        return True
+    if not user.role_id:
+        return False
+    return (
+        db.query(RolePermission)
+        .join(Permission)
+        .filter(
+            RolePermission.role_id == user.role_id,
+            Permission.name == permission_name,
+        )
+        .first()
+        is not None
+    )
+
+
+def _require_schedule_read(db: Session, user: User) -> User:
+    if not _has_permission(db, user, "schedule_type.read"):
+        raise HTTPException(403, "Permission denied: schedule_type.read")
+    return user
+
+
+def _require_schedule_manage(db: Session, user: User) -> User:
+    if not _has_permission(db, user, "schedule_type.manage"):
+        raise HTTPException(403, "Permission denied: schedule_type.manage")
+    return user
+
+
+# ---------------------------------------------------------------------------
+# Schedule Type CRUD
+# ---------------------------------------------------------------------------
+
+@router.get(
+    "/",
+    response_model=List[ScheduleTypeResponse],
+    summary="List all schedule types",
+)
+def list_schedule_types(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    _require_schedule_read(db, current_user)
+    return db.query(ScheduleType).all()
+
+
+@router.post(
+    "/",
+    response_model=ScheduleTypeResponse,
+    summary="Create a schedule type",
+)
+def create_schedule_type(
+    payload: ScheduleTypeCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    _require_schedule_manage(db, current_user)
+
+    existing = db.query(ScheduleType).filter(ScheduleType.name == payload.name).first()
+    if existing:
+        raise HTTPException(409, f"Schedule type '{payload.name}' already exists")
+
+    st = ScheduleType(
+        name=payload.name,
+        work_from=payload.work_from,
+        work_to=payload.work_to,
+        entertainment_from=payload.entertainment_from,
+        entertainment_to=payload.entertainment_to,
+    )
+    db.add(st)
+    db.commit()
+    db.refresh(st)
+    return st
+
+
+@router.patch(
+    "/{schedule_type_id}",
+    response_model=ScheduleTypeResponse,
+    summary="Update a schedule type",
+)
+def update_schedule_type(
+    schedule_type_id: int,
+    payload: ScheduleTypeUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    _require_schedule_manage(db, current_user)
+
+    st = db.query(ScheduleType).filter(ScheduleType.id == schedule_type_id).first()
+    if not st:
+        raise HTTPException(404, "Schedule type not found")
+
+    for field, value in payload.model_dump(exclude_unset=True).items():
+        setattr(st, field, value)
+
+    db.commit()
+    db.refresh(st)
+    return st
+
+
+@router.delete(
+    "/{schedule_type_id}",
+    summary="Delete a schedule type",
+)
+def delete_schedule_type(
+    schedule_type_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    _require_schedule_manage(db, current_user)
+
+    st = db.query(ScheduleType).filter(ScheduleType.id == schedule_type_id).first()
+    if not st:
+        raise HTTPException(404, "Schedule type not found")
+
+    # Check if any agents are using this schedule type
+    agents_using = db.query(Agent).filter(Agent.schedule_type_id == schedule_type_id).count()
+    if agents_using > 0:
+        raise HTTPException(
+            409,
+            f"Cannot delete: {agents_using} agent(s) are assigned to this schedule type",
+        )
+
+    db.delete(st)
+    db.commit()
+    return {"ok": True, "deleted": schedule_type_id}
+
+
+# ---------------------------------------------------------------------------
+# Agent schedule type assignment (agent-facing, uses X-Agent-ID header)
+# ---------------------------------------------------------------------------
+
+@router.get(
+    "/agent/me",
+    response_model=ScheduleTypeResponse | None,
+    summary="Get my schedule type",
+)
+def get_my_schedule_type(
+    x_agent_id: str = Header(..., alias="X-Agent-ID"),
+    x_claw_identifier: str = Header(..., alias="X-Claw-Identifier"),
+    db: Session = Depends(get_db),
+):
+    agent = (
+        db.query(Agent)
+        .filter(Agent.agent_id == x_agent_id, Agent.claw_identifier == x_claw_identifier)
+        .first()
+    )
+    if not agent:
+        raise HTTPException(404, "Agent not found")
+
+    if not agent.schedule_type_id:
+        return None
+
+    return db.query(ScheduleType).filter(ScheduleType.id == agent.schedule_type_id).first()
+
+
+@router.put(
+    "/agent/{agent_id}/assign",
+    summary="Assign a schedule type to an agent",
+)
+def assign_schedule_type(
+    agent_id: str,
+    payload: AgentScheduleTypeAssign,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    _require_schedule_manage(db, current_user)
+
+    agent = db.query(Agent).filter(Agent.agent_id == agent_id).first()
+    if not agent:
+        raise HTTPException(404, f"Agent '{agent_id}' not found")
+
+    st = db.query(ScheduleType).filter(ScheduleType.name == payload.schedule_type_name).first()
+    if not st:
+        raise HTTPException(404, f"Schedule type '{payload.schedule_type_name}' not found")
+
+    agent.schedule_type_id = st.id
+    db.commit()
+    return {"ok": True, "agent_id": agent_id, "schedule_type": st.name}

app/main.py

@@ -63,6 +63,7 @@ from app.api.routers.proposes import router as proposes_router  # legacy compat
 from app.api.routers.milestone_actions import router as milestone_actions_router
 from app.api.routers.meetings import router as meetings_router
 from app.api.routers.essentials import router as essentials_router
+from app.api.routers.schedule_type import router as schedule_type_router
 from app.api.routers.calendar import router as calendar_router
 
 app.include_router(auth_router)
@@ -80,6 +81,7 @@ app.include_router(proposes_router)  # legacy compat
 app.include_router(milestone_actions_router)
 app.include_router(meetings_router)
 app.include_router(essentials_router)
+app.include_router(schedule_type_router)
 app.include_router(calendar_router)
 
 
@@ -363,6 +365,10 @@ def _migrate_schema():
         if _has_table(db, "time_slots") and not _has_column(db, "time_slots", "wakeup_sent_at"):
             db.execute(text("ALTER TABLE time_slots ADD COLUMN wakeup_sent_at DATETIME NULL"))
 
+        # --- agents: add schedule_type_id FK ---
+        if _has_table(db, "agents") and not _has_column(db, "agents", "schedule_type_id"):
+            db.execute(text("ALTER TABLE agents ADD COLUMN schedule_type_id INTEGER NULL"))
+
         db.commit()
     except Exception as e:
         db.rollback()
@@ -397,7 +403,7 @@ def _sync_default_user_roles(db):
 @app.on_event("startup")
 def startup():
     from app.core.config import Base, engine, SessionLocal
-    from app.models import models, webhook, apikey, activity, milestone, notification, worklog, monitor, role_permission, task, support, meeting, proposal, propose, essential, agent, calendar, minimum_workload
+    from app.models import models, webhook, apikey, activity, milestone, notification, worklog, monitor, role_permission, task, support, meeting, proposal, propose, essential, agent, calendar, minimum_workload, schedule_type
     Base.metadata.create_all(bind=engine)
     _migrate_schema()
 

app/models/agent.py

@@ -131,6 +131,15 @@ class Agent(Base):
         comment="rate_limit | billing — why the agent is exhausted",
     )
 
+    # -- schedule type ------------------------------------------------------
+
+    schedule_type_id = Column(
+        Integer,
+        ForeignKey("schedule_types.id"),
+        nullable=True,
+        comment="FK to schedule_types — defines work/entertainment periods",
+    )
+
     # -- timestamps ---------------------------------------------------------
 
     created_at = Column(DateTime(timezone=True), server_default=func.now())
@@ -138,3 +147,4 @@ class Agent(Base):
     # -- relationships ------------------------------------------------------
 
     user = relationship("User", back_populates="agent", uselist=False)
+    schedule_type = relationship("ScheduleType", lazy="joined")

app/models/schedule_type.py

@@ -0,0 +1,52 @@
+"""ScheduleType model — defines work/entertainment time periods.
+
+Each ScheduleType defines the daily work and entertainment windows.
+Agents reference a schedule_type to know when they should be working
+vs when they can engage in entertainment activities.
+"""
+
+from sqlalchemy import Column, Integer, String, DateTime
+from sqlalchemy.sql import func
+from app.core.config import Base
+
+
+class ScheduleType(Base):
+    """Work/entertainment period definition."""
+
+    __tablename__ = "schedule_types"
+
+    id = Column(Integer, primary_key=True, index=True)
+
+    name = Column(
+        String(64),
+        nullable=False,
+        unique=True,
+        comment="Human-readable schedule type name (e.g., 'standard', 'night-shift')",
+    )
+
+    work_from = Column(
+        Integer,
+        nullable=False,
+        comment="Work period start hour (0-23, UTC)",
+    )
+
+    work_to = Column(
+        Integer,
+        nullable=False,
+        comment="Work period end hour (0-23, UTC)",
+    )
+
+    entertainment_from = Column(
+        Integer,
+        nullable=False,
+        comment="Entertainment period start hour (0-23, UTC)",
+    )
+
+    entertainment_to = Column(
+        Integer,
+        nullable=False,
+        comment="Entertainment period end hour (0-23, UTC)",
+    )
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

app/schemas/schedule_type.py

@@ -0,0 +1,36 @@
+"""Schemas for ScheduleType CRUD."""
+
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class ScheduleTypeCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=64)
+    work_from: int = Field(..., ge=0, le=23)
+    work_to: int = Field(..., ge=0, le=23)
+    entertainment_from: int = Field(..., ge=0, le=23)
+    entertainment_to: int = Field(..., ge=0, le=23)
+
+
+class ScheduleTypeUpdate(BaseModel):
+    name: Optional[str] = Field(None, min_length=1, max_length=64)
+    work_from: Optional[int] = Field(None, ge=0, le=23)
+    work_to: Optional[int] = Field(None, ge=0, le=23)
+    entertainment_from: Optional[int] = Field(None, ge=0, le=23)
+    entertainment_to: Optional[int] = Field(None, ge=0, le=23)
+
+
+class ScheduleTypeResponse(BaseModel):
+    id: int
+    name: str
+    work_from: int
+    work_to: int
+    entertainment_from: int
+    entertainment_to: int
+
+    class Config:
+        from_attributes = True
+
+
+class AgentScheduleTypeAssign(BaseModel):
+    schedule_type_name: str = Field(..., description="Name of the schedule type to assign")