fix(calendar): /agent/status idempotent + 409 on bad transition

Same-state transition was 500 (transition_to_busy asserts current=IDLE).
Now: short-circuit identical target → 200 no_change=true. Any other
state-machine violation surfaces as 409 with the actual error message
instead of generic 500.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

app/api/routers/calendar.py

@@ -21,6 +21,11 @@ from app.core.config import get_db
 from app.models.calendar import SchedulePlan, SlotStatus, SlotType, TimeSlot
 from app.models.models import User
 from app.models.agent import Agent, AgentStatus, ExhaustReason
+from app.models.schedule_type import ScheduleType
+from app.services.special_slot_materialiser import (
+    materialise_special_slots_for_claw,
+    materialise_special_slots_for_user,
+)
 from app.schemas.calendar import (
     AgentHeartbeatResponse,
     AgentStatusUpdateRequest,
@@ -47,6 +52,7 @@ from app.schemas.calendar import (
 )
 from app.services.agent_heartbeat import get_pending_slots_for_agent
 from app.services.agent_status import (
+    AgentStatusError,
     record_heartbeat,
     transition_to_busy,
     transition_to_idle,
@@ -143,6 +149,8 @@ def _slot_to_response(slot: TimeSlot) -> TimeSlotResponse:
         priority=slot.priority,
         status=slot.status.value if hasattr(slot.status, "value") else str(slot.status),
         plan_id=slot.plan_id,
+        is_admin_locked=bool(getattr(slot, "is_admin_locked", False)),
+        special_slot_id=getattr(slot, "special_slot_id", None),
         created_at=slot.created_at,
         updated_at=slot.updated_at,
     )
@@ -168,6 +176,48 @@ def create_slot(
     """
     target_date = payload.date or date_type.today()
 
+    # --- Maintenance-window guard ---
+    # Non-`system` slots may not be placed inside the schedule_type's
+    # 1-hour maintenance window. The window is admin-territory, reserved
+    # for materialised special slots from `schedule_type_special_slots`.
+    # `system` slot_type is itself reserved server-side (the materialiser
+    # is the only legitimate caller) — refuse it here outright so the
+    # public API cannot manufacture a fake admin-locked slot.
+    if payload.slot_type == SlotType.SYSTEM:
+        raise HTTPException(
+            status_code=422,
+            detail=(
+                "slot_type='system' is reserved for schedule_type special slots "
+                "and cannot be created via this endpoint"
+            ),
+        )
+    _agent_for_user = (
+        db.query(Agent).filter(Agent.user_id == current_user.id).first()
+    )
+    if _agent_for_user and _agent_for_user.schedule_type_id:
+        st = (
+            db.query(ScheduleType)
+            .filter(ScheduleType.id == _agent_for_user.schedule_type_id)
+            .first()
+        )
+        if st and _scheduled_inside_window(
+            payload.scheduled_at,
+            payload.estimated_duration,
+            st.maintenance_from,
+            st.maintenance_to,
+        ):
+            mf_h, mf_m = divmod(st.maintenance_from, 60)
+            mt_h, mt_m = divmod(st.maintenance_to, 60)
+            raise HTTPException(
+                status_code=422,
+                detail=(
+                    f"slot at {payload.scheduled_at} duration {payload.estimated_duration}min "
+                    f"intersects the maintenance window "
+                    f"{mf_h:02d}:{mf_m:02d}-{mt_h:02d}:{mt_m:02d} UTC of "
+                    f"schedule_type '{st.name}' — that window is admin-reserved"
+                ),
+            )
+
     # --- Overlap check (hard reject) ---
     conflicts = check_overlap_for_create(
         db,
@@ -236,6 +286,8 @@ def _real_slot_to_item(slot: TimeSlot) -> CalendarSlotItem:
         priority=slot.priority,
         status=slot.status.value if hasattr(slot.status, "value") else str(slot.status),
         plan_id=slot.plan_id,
+        is_admin_locked=bool(getattr(slot, "is_admin_locked", False)),
+        special_slot_id=getattr(slot, "special_slot_id", None),
         created_at=slot.created_at,
         updated_at=slot.updated_at,
     )
@@ -289,7 +341,47 @@ def _require_agent(db: Session, agent_id: str, claw_identifier: str) -> Agent:
     return agent
 
 
+def _scheduled_inside_window(
+    scheduled_at,
+    estimated_duration_minutes: int,
+    window_from_min: int,
+    window_to_min: int,
+) -> bool:
+    """True if [scheduled_at, scheduled_at+duration] intersects [from, to).
+
+    Window bounds are minutes-since-UTC-midnight (0-1439). Handles the
+    case where the window crosses UTC midnight (e.g. 23:30→01:00).
+    """
+    start_min = scheduled_at.hour * 60 + scheduled_at.minute
+    end_min = start_min + max(estimated_duration_minutes, 1)
+    if window_to_min > window_from_min:
+        # normal same-day window
+        return start_min < window_to_min and end_min > window_from_min
+    # wrap-around: window = [from..1440) ∪ [0..to)
+    return (start_min < 1440 and end_min > window_from_min) or end_min > window_to_min
+
+
+# Admin-locked special slots accept only these agent-driven status
+# transitions; movement / cancellation / arbitrary status edits are
+# rejected because the schedule_type owner is the source of truth.
+_ADMIN_LOCKED_ALLOWED_STATUSES = {
+    SlotStatusEnum.ONGOING,
+    SlotStatusEnum.PAUSED,
+    SlotStatusEnum.FINISHED,
+    SlotStatusEnum.ABORTED,
+}
+
+
 def _apply_agent_slot_update(slot: TimeSlot, payload: SlotAgentUpdate) -> None:
+    if getattr(slot, "is_admin_locked", False):
+        if payload.status not in _ADMIN_LOCKED_ALLOWED_STATUSES:
+            raise HTTPException(
+                status_code=423,
+                detail=(
+                    f"slot {slot.id} is admin-locked (special slot); only "
+                    f"ongoing/paused/finished/aborted are allowed via agent-update"
+                ),
+            )
     slot.status = payload.status.value
     if payload.started_at is not None:
         slot.started_at = payload.started_at
@@ -377,6 +469,12 @@ def sync_schedules(
     """
     today = date_type.today()
 
+    # Materialise today's special slots for every agent on this claw
+    # before reading. This is idempotent — re-runs against an already-
+    # materialised (agent, date, template) are no-ops. Plugin's runSync
+    # picks them up like any other slot via the normal real_slots query.
+    materialise_special_slots_for_claw(db, x_claw_identifier, today, commit=True)
+
     # Find all agents on this claw instance
     agents = (
         db.query(Agent)
@@ -464,6 +562,29 @@ def agent_update_virtual_slot(
     return TimeSlotEditResponse(slot=_slot_to_response(slot), warnings=[])
 
 
+@router.get(
+    "/agent/status",
+    summary="Read an agent's current runtime status (no side effects)",
+)
+def get_agent_status(
+    agent_id: str = Query(..., description="Target agent_id"),
+    x_claw_identifier: str = Header(..., alias="X-Claw-Identifier"),
+    db: Session = Depends(get_db),
+):
+    """Return `{agent_id, status}` so callers (Fabric.OpenclawPlugin's
+    triage on-call gate, etc.) can decide whether the agent is currently
+    eligible without flipping their state.
+
+    No-op for unknown agents — returns 404 with `{detail: 'Agent not
+    found'}` so the caller can decide whether to fail-open or fail-closed.
+    """
+    agent = _require_agent(db, agent_id, x_claw_identifier)
+    return {
+        "agent_id": agent.agent_id,
+        "status": agent.status.value if hasattr(agent.status, 'value') else str(agent.status),
+    }
+
+
 @router.post(
     "/agent/status",
     summary="Update agent runtime status from plugin",
@@ -474,19 +595,31 @@ def update_agent_status(
 ):
     agent = _require_agent(db, payload.agent_id, payload.claw_identifier)
     target = (payload.status or '').lower().strip()
-    if target == AgentStatus.IDLE.value:
-        transition_to_idle(db, agent)
-    elif target == AgentStatus.BUSY.value:
-        transition_to_busy(db, agent, slot_type=SlotType.WORK)
-    elif target == AgentStatus.ON_CALL.value:
-        transition_to_busy(db, agent, slot_type=SlotType.ON_CALL)
-    elif target == AgentStatus.OFFLINE.value:
-        transition_to_offline(db, agent)
-    elif target == AgentStatus.EXHAUSTED.value:
-        reason = ExhaustReason.BILLING if payload.exhaust_reason == 'billing' else ExhaustReason.RATE_LIMIT
-        transition_to_exhausted(db, agent, reason=reason, recovery_at=payload.recovery_at)
-    else:
-        raise HTTPException(status_code=400, detail="Unsupported agent status")
+    # Idempotent same-state transition: a 'busy → busy' request is a
+    # no-op rather than a 500. Lets plugin status gates / cli `--set`
+    # be safe to fire-and-forget without first reading current state.
+    current = agent.status.value if hasattr(agent.status, 'value') else str(agent.status)
+    if current == target:
+        return {"ok": True, "agent_id": agent.agent_id, "status": current, "no_change": True}
+    try:
+        if target == AgentStatus.IDLE.value:
+            transition_to_idle(db, agent)
+        elif target == AgentStatus.BUSY.value:
+            transition_to_busy(db, agent, slot_type=SlotType.WORK)
+        elif target == AgentStatus.ON_CALL.value:
+            transition_to_busy(db, agent, slot_type=SlotType.ON_CALL)
+        elif target == AgentStatus.OFFLINE.value:
+            transition_to_offline(db, agent)
+        elif target == AgentStatus.EXHAUSTED.value:
+            reason = ExhaustReason.BILLING if payload.exhaust_reason == 'billing' else ExhaustReason.RATE_LIMIT
+            transition_to_exhausted(db, agent, reason=reason, recovery_at=payload.recovery_at)
+        else:
+            raise HTTPException(status_code=400, detail="Unsupported agent status")
+    except AgentStatusError as e:
+        # State-machine violation (e.g. busy → busy via wrong precondition)
+        # → 409 with the rejected transition explained, instead of a 500.
+        db.rollback()
+        raise HTTPException(status_code=409, detail=str(e))
     db.commit()
     return {"ok": True, "agent_id": agent.agent_id, "status": agent.status.value if hasattr(agent.status, 'value') else str(agent.status)}
 
@@ -514,6 +647,10 @@ def get_calendar_day(
     """
     target_date = date or date_type.today()
 
+    # Materialise today's special slots for this user before reading,
+    # so the day-view returns them alongside any user-created slots.
+    materialise_special_slots_for_user(db, current_user.id, target_date, commit=True)
+
     # 1. Fetch real slots for the day
     real_slots = (
         db.query(TimeSlot)
@@ -589,6 +726,20 @@ def edit_real_slot(
     if slot is None:
         raise HTTPException(status_code=404, detail="Slot not found")
 
+    # --- Admin-locked guard ---
+    # Special slots materialised from a schedule_type template are
+    # admin-owned; agents may complete/abort/pause/resume via the
+    # plugin-facing agent-update endpoint but cannot edit time/type/
+    # duration/event-data via this user-facing edit endpoint.
+    if getattr(slot, "is_admin_locked", False):
+        raise HTTPException(
+            status_code=423,
+            detail=(
+                f"slot {slot.id} is admin-locked (materialised from a special "
+                f"slot template); only the schedule_type owner can edit it"
+            ),
+        )
+
     # --- Past-slot guard ---
     try:
         guard_edit_real_slot(db, slot)
@@ -756,6 +907,16 @@ def cancel_real_slot(
     if slot is None:
         raise HTTPException(status_code=404, detail="Slot not found")
 
+    # --- Admin-locked guard ---
+    if getattr(slot, "is_admin_locked", False):
+        raise HTTPException(
+            status_code=423,
+            detail=(
+                f"slot {slot.id} is admin-locked (materialised from a special "
+                f"slot template); only the schedule_type owner can cancel it"
+            ),
+        )
+
     # --- Past-slot guard ---
     try:
         guard_cancel_real_slot(db, slot)

app/api/routers/schedule_type.py

@@ -9,7 +9,7 @@ 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.api.deps import get_current_user_or_apikey
 from app.models.models import User
 from app.models.agent import Agent
 from app.models.schedule_type import ScheduleType
@@ -57,6 +57,18 @@ def _require_schedule_manage(db: Session, user: User) -> User:
     return user
 
 
+def _attach_derived(st: ScheduleType) -> ScheduleType:
+    """Attach derived fields (maintenance_duration_minutes) so the
+    pydantic ScheduleTypeResponse picks them up via from_attributes.
+
+    Pydantic with from_attributes reads attributes off the ORM object;
+    setting a transient attr here avoids having to convert through dict.
+    """
+    if st is not None:
+        st.maintenance_duration_minutes = st.compute_maintenance_duration()
+    return st
+
+
 # ---------------------------------------------------------------------------
 # Schedule Type CRUD
 # ---------------------------------------------------------------------------
@@ -68,10 +80,10 @@ def _require_schedule_manage(db: Session, user: User) -> User:
 )
 def list_schedule_types(
     db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(get_current_user_or_apikey),
 ):
     _require_schedule_read(db, current_user)
-    return db.query(ScheduleType).all()
+    return [_attach_derived(st) for st in db.query(ScheduleType).all()]
 
 
 @router.post(
@@ -82,7 +94,7 @@ def list_schedule_types(
 def create_schedule_type(
     payload: ScheduleTypeCreate,
     db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(get_current_user_or_apikey),
 ):
     _require_schedule_manage(db, current_user)
 
@@ -96,11 +108,13 @@ def create_schedule_type(
         work_to=payload.work_to,
         entertainment_from=payload.entertainment_from,
         entertainment_to=payload.entertainment_to,
+        maintenance_from=payload.maintenance_from,
+        maintenance_to=payload.maintenance_to,
     )
     db.add(st)
     db.commit()
     db.refresh(st)
-    return st
+    return _attach_derived(st)
 
 
 @router.patch(
@@ -112,7 +126,7 @@ def update_schedule_type(
     schedule_type_id: int,
     payload: ScheduleTypeUpdate,
     db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(get_current_user_or_apikey),
 ):
     _require_schedule_manage(db, current_user)
 
@@ -120,12 +134,23 @@ def update_schedule_type(
     if not st:
         raise HTTPException(404, "Schedule type not found")
 
-    for field, value in payload.model_dump(exclude_unset=True).items():
+    update_fields = payload.model_dump(exclude_unset=True)
+    for field, value in update_fields.items():
         setattr(st, field, value)
 
+    # Re-validate maintenance after merge (partial updates can put the row
+    # into an invalid window combo that the pydantic schema couldn't catch
+    # because it only saw one field).
+    from app.schemas.schedule_type import _validate_maintenance_window
+    try:
+        _validate_maintenance_window(st.maintenance_from, st.maintenance_to)
+    except ValueError as e:
+        db.rollback()
+        raise HTTPException(422, str(e))
+
     db.commit()
     db.refresh(st)
-    return st
+    return _attach_derived(st)
 
 
 @router.delete(
@@ -135,7 +160,7 @@ def update_schedule_type(
 def delete_schedule_type(
     schedule_type_id: int,
     db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(get_current_user_or_apikey),
 ):
     _require_schedule_manage(db, current_user)
 
@@ -181,7 +206,8 @@ def get_my_schedule_type(
     if not agent.schedule_type_id:
         return None
 
-    return db.query(ScheduleType).filter(ScheduleType.id == agent.schedule_type_id).first()
+    st = db.query(ScheduleType).filter(ScheduleType.id == agent.schedule_type_id).first()
+    return _attach_derived(st) if st else None
 
 
 @router.put(
@@ -192,7 +218,7 @@ def assign_schedule_type(
     agent_id: str,
     payload: AgentScheduleTypeAssign,
     db: Session = Depends(get_db),
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(get_current_user_or_apikey),
 ):
     _require_schedule_manage(db, current_user)
 

app/api/routers/schedule_type_special_slot.py

@@ -0,0 +1,226 @@
+"""Special-slot CRUD for a ScheduleType (admin-only).
+
+A "special slot" is a recurring slot template tied to a ScheduleType.
+The system materialises one `time_slots` row per agent on that
+schedule_type per date, scheduled inside the schedule_type's
+maintenance window. Materialised rows are `is_admin_locked=true` —
+agents can complete / abort / pause / resume them but cannot move
+or cancel them.
+
+All endpoints require `schedule_type.manage` (admin auto-grants).
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.core.config import get_db
+from app.api.deps import get_current_user_or_apikey
+from app.models.models import User
+from app.models.role_permission import Permission, RolePermission
+from app.models.schedule_type import ScheduleType
+from app.models.schedule_type_special_slot import ScheduleTypeSpecialSlot
+from app.schemas.schedule_type_special_slot import (
+    SpecialSlotCreate,
+    SpecialSlotUpdate,
+    SpecialSlotResponse,
+)
+
+
+router = APIRouter(prefix="/schedule-types", tags=["ScheduleTypes"])
+
+
+# ---------------------------------------------------------------------------
+# Permission helpers — mirror schedule_type.py's local helpers so this router
+# doesn't have to depend on internal symbols of the other router.
+# ---------------------------------------------------------------------------
+
+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_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
+
+
+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 _fetch_schedule_type(db: Session, schedule_type_id: int) -> ScheduleType:
+    st = db.query(ScheduleType).filter(ScheduleType.id == schedule_type_id).first()
+    if not st:
+        raise HTTPException(404, f"ScheduleType {schedule_type_id} not found")
+    return st
+
+
+def _validate_fits_window(
+    minute_in_window: int,
+    estimated_duration: int,
+    maintenance_duration_minutes: int,
+) -> None:
+    """Reject special slots that wouldn't fit inside the parent's maintenance window."""
+    if minute_in_window + estimated_duration > maintenance_duration_minutes:
+        raise HTTPException(
+            422,
+            (
+                f"special slot does not fit in maintenance window: "
+                f"minute_in_window={minute_in_window} + "
+                f"estimated_duration={estimated_duration} > "
+                f"maintenance window {maintenance_duration_minutes}min"
+            ),
+        )
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+@router.get(
+    "/{schedule_type_id}/special-slots",
+    response_model=List[SpecialSlotResponse],
+    summary="List special slots for a schedule type",
+)
+def list_special_slots(
+    schedule_type_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user_or_apikey),
+):
+    _require_schedule_read(db, current_user)
+    _fetch_schedule_type(db, schedule_type_id)
+    return (
+        db.query(ScheduleTypeSpecialSlot)
+        .filter(ScheduleTypeSpecialSlot.schedule_type_id == schedule_type_id)
+        .order_by(
+            ScheduleTypeSpecialSlot.minute_in_window.asc(),
+            ScheduleTypeSpecialSlot.id.asc(),
+        )
+        .all()
+    )
+
+
+@router.post(
+    "/{schedule_type_id}/special-slots",
+    response_model=SpecialSlotResponse,
+    summary="Create a special slot for a schedule type (admin)",
+)
+def create_special_slot(
+    schedule_type_id: int,
+    payload: SpecialSlotCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user_or_apikey),
+):
+    _require_schedule_manage(db, current_user)
+    st = _fetch_schedule_type(db, schedule_type_id)
+    _validate_fits_window(payload.minute_in_window, payload.estimated_duration, st.compute_maintenance_duration())
+
+    dup = (
+        db.query(ScheduleTypeSpecialSlot)
+        .filter(
+            ScheduleTypeSpecialSlot.schedule_type_id == schedule_type_id,
+            ScheduleTypeSpecialSlot.name == payload.name,
+        )
+        .first()
+    )
+    if dup:
+        raise HTTPException(
+            409,
+            f"special slot '{payload.name}' already exists for schedule_type {schedule_type_id}",
+        )
+
+    slot = ScheduleTypeSpecialSlot(
+        schedule_type_id=schedule_type_id,
+        name=payload.name,
+        description=payload.description,
+        minute_in_window=payload.minute_in_window,
+        estimated_duration=payload.estimated_duration,
+        priority=payload.priority,
+        event_data=payload.event_data,
+        is_active=payload.is_active,
+        created_by_user_id=current_user.id,
+    )
+    db.add(slot)
+    db.commit()
+    db.refresh(slot)
+    return slot
+
+
+@router.patch(
+    "/{schedule_type_id}/special-slots/{slot_id}",
+    response_model=SpecialSlotResponse,
+    summary="Update a special slot (admin)",
+)
+def update_special_slot(
+    schedule_type_id: int,
+    slot_id: int,
+    payload: SpecialSlotUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user_or_apikey),
+):
+    _require_schedule_manage(db, current_user)
+    slot = (
+        db.query(ScheduleTypeSpecialSlot)
+        .filter(
+            ScheduleTypeSpecialSlot.id == slot_id,
+            ScheduleTypeSpecialSlot.schedule_type_id == schedule_type_id,
+        )
+        .first()
+    )
+    if not slot:
+        raise HTTPException(404, "Special slot not found")
+
+    update_fields = payload.model_dump(exclude_unset=True)
+    next_min = update_fields.get("minute_in_window", slot.minute_in_window)
+    next_dur = update_fields.get("estimated_duration", slot.estimated_duration)
+    parent = _fetch_schedule_type(db, schedule_type_id)
+    _validate_fits_window(next_min, next_dur, parent.compute_maintenance_duration())
+
+    for field, value in update_fields.items():
+        setattr(slot, field, value)
+    db.commit()
+    db.refresh(slot)
+    return slot
+
+
+@router.delete(
+    "/{schedule_type_id}/special-slots/{slot_id}",
+    summary="Delete a special slot (admin)",
+)
+def delete_special_slot(
+    schedule_type_id: int,
+    slot_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user_or_apikey),
+):
+    _require_schedule_manage(db, current_user)
+    slot = (
+        db.query(ScheduleTypeSpecialSlot)
+        .filter(
+            ScheduleTypeSpecialSlot.id == slot_id,
+            ScheduleTypeSpecialSlot.schedule_type_id == schedule_type_id,
+        )
+        .first()
+    )
+    if not slot:
+        raise HTTPException(404, "Special slot not found")
+    db.delete(slot)
+    db.commit()
+    return {"ok": True, "deleted": slot_id}

app/api/routers/users.py

@@ -221,6 +221,71 @@ def update_user(
     return _user_response(user)
 
 
+@router.patch("/{identifier}/bind-agent", response_model=schemas.UserResponse)
+def bind_agent(
+    identifier: str,
+    payload: schemas.UserBindAgentRequest,
+    db: Session = Depends(get_db),
+    _: models.User = Depends(require_account_creator),
+):
+    """Bind an existing user to (agent_id, claw_identifier).
+
+    Backfill path for users that were created via `hf user create` before
+    the cli supported `--agent-id` / `--claw-identifier` flags. Creates
+    the `agents` row that should have been written at user-create time.
+
+    Idempotent: if the user is already bound to the same
+    (agent_id, claw_identifier), returns the user unchanged (200, no-op).
+
+    Rejects (409) if:
+      - the user is bound to a DIFFERENT (agent_id, claw_identifier)
+      - the requested agent_id is already in use by another user
+
+    Permission: account.create (admin auto-grants) — same gate as
+    POST /users so the surface stays symmetric.
+    """
+    user = _find_user_by_id_or_username(db, identifier)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    existing_agent_for_user = db.query(Agent).filter(Agent.user_id == user.id).first()
+    if existing_agent_for_user:
+        if (
+            existing_agent_for_user.agent_id == payload.agent_id
+            and existing_agent_for_user.claw_identifier == payload.claw_identifier
+        ):
+            # idempotent re-bind
+            return _user_response(user)
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"User '{user.username}' is already bound to agent "
+                f"'{existing_agent_for_user.agent_id}' on claw "
+                f"'{existing_agent_for_user.claw_identifier}'"
+            ),
+        )
+
+    existing_for_agent_id = (
+        db.query(Agent).filter(Agent.agent_id == payload.agent_id).first()
+    )
+    if existing_for_agent_id:
+        raise HTTPException(
+            status_code=409,
+            detail=f"agent_id '{payload.agent_id}' already in use by another user",
+        )
+
+    db.add(
+        Agent(
+            user_id=user.id,
+            agent_id=payload.agent_id,
+            claw_identifier=payload.claw_identifier,
+        )
+    )
+    db.commit()
+    db.refresh(user)
+    return _user_response(user)
+
+
 _BUILTIN_USERNAMES = {"acc-mgr", DELETED_USER_USERNAME}
 
 

app/main.py

@@ -78,6 +78,7 @@ 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.schedule_type_special_slot import router as schedule_type_special_slot_router
 from app.api.routers.calendar import router as calendar_router
 from app.api.routers.oidc import router as oidc_router
 
@@ -98,6 +99,7 @@ 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(schedule_type_special_slot_router)
 app.include_router(calendar_router)
 
 
@@ -397,6 +399,63 @@ def _migrate_schema():
         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"))
 
+        # --- schedule_types: add maintenance_from / maintenance_to ---
+        # Default 8:00–9:00 UTC for existing rows; the maintenance
+        # duration invariant (1-180min) is enforced at the schema
+        # level for any NEW rows by ScheduleTypeCreate validator.
+        if _has_table(db, "schedule_types"):
+            if not _has_column(db, "schedule_types", "maintenance_from"):
+                db.execute(text(
+                    "ALTER TABLE schedule_types ADD COLUMN maintenance_from INT NOT NULL DEFAULT 8"
+                ))
+            if not _has_column(db, "schedule_types", "maintenance_to"):
+                db.execute(text(
+                    "ALTER TABLE schedule_types ADD COLUMN maintenance_to INT NOT NULL DEFAULT 9"
+                ))
+
+            # --- minutes-since-midnight migration (PR #21+) ---
+            # The 6 schedule_type window columns used to hold *hours*
+            # (0-23). PR #21 changed semantics to *minutes since UTC
+            # midnight* (0-1439). Detect the legacy regime by checking
+            # if ANY row has all 6 values ≤ 23 — if so, multiply each
+            # by 60 to convert. Idempotent: post-conversion values are
+            # all ≥ 0 and usually well above 23, so guard never fires
+            # twice.
+            row = db.execute(text(
+                "SELECT MAX(GREATEST(work_from, work_to, entertainment_from, entertainment_to, maintenance_from, maintenance_to)) AS m "
+                "FROM schedule_types"
+            )).fetchone()
+            if row is not None and row.m is not None and row.m <= 23:
+                db.execute(text(
+                    "UPDATE schedule_types SET "
+                    "  work_from = work_from * 60, "
+                    "  work_to = work_to * 60, "
+                    "  entertainment_from = entertainment_from * 60, "
+                    "  entertainment_to = entertainment_to * 60, "
+                    "  maintenance_from = maintenance_from * 60, "
+                    "  maintenance_to = maintenance_to * 60"
+                ))
+
+        # --- time_slots: admin-locked + special_slot pointer ---
+        if _has_table(db, "time_slots"):
+            if not _has_column(db, "time_slots", "is_admin_locked"):
+                db.execute(text(
+                    "ALTER TABLE time_slots ADD COLUMN is_admin_locked TINYINT(1) NOT NULL DEFAULT 0"
+                ))
+            if not _has_column(db, "time_slots", "special_slot_id"):
+                db.execute(text(
+                    "ALTER TABLE time_slots ADD COLUMN special_slot_id INTEGER NULL"
+                ))
+                # Index for the materialiser's idempotency lookup
+                db.execute(text(
+                    "CREATE INDEX idx_time_slots_special_slot_id ON time_slots (special_slot_id)"
+                ))
+
+        # --- schedule_type_special_slots: create-table is handled by
+        # Base.metadata.create_all on first boot; no migration needed here
+        # because there is no legacy table to evolve. Future schema bumps
+        # to that table go in this block.
+
         db.commit()
     except Exception as e:
         db.rollback()
@@ -431,7 +490,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, schedule_type, oidc_settings
+    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, schedule_type_special_slot, oidc_settings
     Base.metadata.create_all(bind=engine)
     _migrate_schema()
 

app/models/calendar.py

@@ -178,11 +178,37 @@ class TimeSlot(Base):
         comment="Source plan if materialized from a SchedulePlan; set NULL on edit/cancel",
     )
 
+    # -----------------------------------------------------------------
+    # Admin-locked slots are materialised from a ScheduleTypeSpecialSlot
+    # template. The agent can complete / abort / pause / resume them but
+    # cannot edit their time, type, duration, or cancel them outright —
+    # the slot exists because admin decided every agent on the parent
+    # schedule_type should run it. See `_apply_agent_slot_update` for
+    # the enforcement.
+    # -----------------------------------------------------------------
+    is_admin_locked = Column(
+        Boolean,
+        nullable=False,
+        server_default="0",
+        comment="True for slots materialised from a schedule_type special slot template.",
+    )
+
+    # Pointer back to the template that materialised this slot. NULL for
+    # all user-created or plan-generated slots. Lets us cascade updates
+    # and surface 'why is this on my calendar' to the agent.
+    special_slot_id = Column(
+        Integer,
+        ForeignKey("schedule_type_special_slots.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
+
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
 
     # relationship ----------------------------------------------------------
     plan = relationship("SchedulePlan", back_populates="materialized_slots")
+    special_slot = relationship("ScheduleTypeSpecialSlot")
 
 
 # ---------------------------------------------------------------------------

app/models/schedule_type.py

@@ -1,17 +1,27 @@
-"""ScheduleType model — defines work/entertainment time periods.
+"""ScheduleType model — defines work/entertainment/maintenance 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.
+Each ScheduleType defines the daily work, entertainment, and maintenance
+windows for agents who reference this type. All bounds are stored as
+**minutes-since-UTC-midnight** (0-1439 inclusive) so half-hour and other
+sub-hour boundaries are exact.
+
+Maintenance window length is variable (1-180 minutes) and admin-owned;
+agent slots cannot intersect it (see `app/api/routers/calendar.py`).
+
+Historical note: pre-PR #21 the columns held *hours* (0-23) and the
+maintenance window was hard-fixed at exactly 1 hour. The additive
+migration in `_migrate_schema()` multiplies legacy values by 60 so
+existing rows convert transparently.
 """
 
 from sqlalchemy import Column, Integer, String, DateTime
+from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
 from app.core.config import Base
 
 
 class ScheduleType(Base):
-    """Work/entertainment period definition."""
+    """Work/entertainment/maintenance period definition."""
 
     __tablename__ = "schedule_types"
 
@@ -24,29 +34,50 @@ class ScheduleType(Base):
         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)",
-    )
+    # Minutes since UTC midnight, 0-1439 inclusive.
+    work_from = Column(Integer, nullable=False, comment="Work period start (minutes since UTC midnight)")
+    work_to = Column(Integer, nullable=False, comment="Work period end (minutes since UTC midnight)")
 
-    work_to = Column(
-        Integer,
-        nullable=False,
-        comment="Work period end hour (0-23, UTC)",
-    )
+    entertainment_from = Column(Integer, nullable=False, comment="Entertainment start (minutes since UTC midnight)")
+    entertainment_to = Column(Integer, nullable=False, comment="Entertainment end (minutes since UTC midnight)")
 
-    entertainment_from = Column(
+    # Maintenance window — admin-owned, variable length (1-180 min).
+    # Default 8:00–9:00 UTC = 480–540 minutes for existing rows.
+    maintenance_from = Column(
         Integer,
         nullable=False,
-        comment="Entertainment period start hour (0-23, UTC)",
+        server_default="480",
+        comment="Maintenance start (minutes since UTC midnight, default 480 = 8:00 UTC).",
     )
-
-    entertainment_to = Column(
+    maintenance_to = Column(
         Integer,
         nullable=False,
-        comment="Entertainment period end hour (0-23, UTC)",
+        server_default="540",
+        comment="Maintenance end (minutes since UTC midnight, default 540 = 9:00 UTC). Duration ((to-from) mod 1440) must be in [1, 180].",
     )
 
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+
+    # relationship ---------------------------------------------------
+    special_slots = relationship(
+        "ScheduleTypeSpecialSlot",
+        back_populates="schedule_type",
+        cascade="all, delete-orphan",
+    )
+
+    # ---------------------------------------------------------------
+    # Convenience methods used by the API layer + materialiser.
+    # ---------------------------------------------------------------
+
+    def compute_maintenance_duration(self) -> int:
+        """Maintenance window length in minutes (handles 23→0 wrap)."""
+        return (self.maintenance_to - self.maintenance_from) % 1440 or 1440
+
+    def window_contains(self, start_min: int, end_min: int, win_from: int, win_to: int) -> bool:
+        """True if [start_min, end_min) intersects [win_from, win_to) (handles wrap)."""
+        # Normalise into [0, 1440) — same logic as the helper in calendar.py.
+        if win_to > win_from:
+            return start_min < win_to and end_min > win_from
+        # wrap window crosses midnight: [win_from..1440) ∪ [0..win_to)
+        return start_min < win_to or end_min > win_from

app/models/schedule_type_special_slot.py

@@ -0,0 +1,116 @@
+"""ScheduleTypeSpecialSlot — admin-managed slot template tied to a ScheduleType.
+
+A "special slot" is a recurring slot template that the system materializes
+into every matching agent's `time_slots` row each day. It exists for tasks
+that admin wants to enforce across an entire schedule type cohort, e.g.:
+
+  * `plan-schedule` — daily planning slot all agents on this type must run
+  * `secret-rotation-window` — security maintenance
+  * `policy-update` — read updated agent policies
+
+Rules:
+  * Only admins (`schedule_type.manage` permission) may create / edit /
+    delete special slots.
+  * The slot's `minute_in_window` offset must place it inside the parent
+    schedule_type's maintenance window (`maintenance_from..maintenance_from+59`).
+  * Materialised `time_slots` rows from a special slot carry
+    `is_admin_locked=true` so the agent-side `PATCH .../agent-update`
+    refuses status/time edits other than complete/abort/pause/resume.
+  * Materialisation produces one `time_slots` row per agent using this
+    schedule_type per date, with `slot_type=system`, `event_type=system_event`,
+    `event_data={"special_slot_id": <id>, "special_slot_name": "<name>",
+                  "source": "schedule_type_special_slot", ...admin-supplied...}`.
+"""
+
+from sqlalchemy import Column, Integer, String, ForeignKey, JSON, DateTime, Boolean, UniqueConstraint
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+
+from app.core.config import Base
+
+
+class ScheduleTypeSpecialSlot(Base):
+    """Admin-managed daily slot template attached to a ScheduleType."""
+
+    __tablename__ = "schedule_type_special_slots"
+    __table_args__ = (
+        # One slot template per (schedule_type, name) so admin can use the
+        # `name` field as a stable, human-readable identifier for the cohort.
+        UniqueConstraint("schedule_type_id", "name", name="uq_special_slot_type_name"),
+    )
+
+    id = Column(Integer, primary_key=True, index=True)
+
+    schedule_type_id = Column(
+        Integer,
+        ForeignKey("schedule_types.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+
+    name = Column(
+        String(64),
+        nullable=False,
+        comment="Short identifier, e.g. 'plan-schedule', 'secret-rotation'",
+    )
+
+    description = Column(
+        String(512),
+        nullable=True,
+        comment="Human-readable note on what this slot is for",
+    )
+
+    minute_in_window = Column(
+        Integer,
+        nullable=False,
+        server_default="0",
+        comment=(
+            "Minute offset (0-59) inside the schedule_type maintenance window. "
+            "The materialised time_slot's scheduled_at becomes "
+            "maintenance_from:minute_in_window:00 UTC."
+        ),
+    )
+
+    estimated_duration = Column(
+        Integer,
+        nullable=False,
+        server_default="15",
+        comment="Duration in minutes. Must fit inside the maintenance window.",
+    )
+
+    priority = Column(
+        Integer,
+        nullable=False,
+        server_default="50",
+        comment="Wake priority — higher value wakes first if multiple slots are due.",
+    )
+
+    event_data = Column(
+        JSON,
+        nullable=True,
+        comment=(
+            "Admin-supplied JSON payload that gets merged into every "
+            "materialised slot's event_data. Use this to pass a workflow "
+            "tag, suggested_workload, or any other context the agent "
+            "should see in its wakeup message."
+        ),
+    )
+
+    is_active = Column(
+        Boolean,
+        nullable=False,
+        server_default="1",
+        comment="Soft-disable without deleting; inactive templates are skipped during materialisation.",
+    )
+
+    created_by_user_id = Column(
+        Integer,
+        ForeignKey("users.id"),
+        nullable=False,
+    )
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+
+    # relationship ---------------------------------------------------
+    schedule_type = relationship("ScheduleType", back_populates="special_slots")

app/schemas/calendar.py

@@ -144,6 +144,8 @@ class TimeSlotResponse(BaseModel):
     priority: int
     status: str
     plan_id: Optional[int] = None
+    is_admin_locked: bool = False
+    special_slot_id: Optional[int] = None
     created_at: Optional[dt_datetime] = None
     updated_at: Optional[dt_datetime] = None
 
@@ -226,6 +228,8 @@ class CalendarSlotItem(BaseModel):
     priority: int
     status: str
     plan_id: Optional[int] = None
+    is_admin_locked: bool = False
+    special_slot_id: Optional[int] = None
     created_at: Optional[dt_datetime] = None
     updated_at: Optional[dt_datetime] = None
 

app/schemas/schedule_type.py

@@ -1,23 +1,67 @@
-"""Schemas for ScheduleType CRUD."""
+"""Schemas for ScheduleType CRUD.
 
-from pydantic import BaseModel, Field
+All `*_from` / `*_to` values are **minutes since UTC midnight** (0-1439).
+A maintenance window of variable length is allowed (1-180 minutes,
+handles 23→0 wrap).
+"""
+
+from pydantic import BaseModel, Field, model_validator
 from typing import Optional
 
 
+_MAX_MIN = 1440  # 24 * 60 — exclusive upper bound
+
+
+def _maintenance_duration(maint_from: int, maint_to: int) -> int:
+    """Maintenance window length in minutes; treats from==to as 24h (invalid)."""
+    return (maint_to - maint_from) % _MAX_MIN or _MAX_MIN
+
+
+def _validate_maintenance_window(maint_from: int, maint_to: int) -> None:
+    dur = _maintenance_duration(maint_from, maint_to)
+    if dur < 1 or dur > 180:
+        raise ValueError(
+            f"maintenance window duration must be in [1, 180] minutes; "
+            f"got {dur} (from={maint_from}, to={maint_to})"
+        )
+
+
+def _validate_minute_field(name: str, value: int) -> None:
+    if value < 0 or value >= _MAX_MIN:
+        raise ValueError(f"{name} must be in [0, {_MAX_MIN}); got {value}")
+
+
 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)
+    work_from: int = Field(..., ge=0, lt=_MAX_MIN, description="Work start (minutes since UTC midnight, 0-1439)")
+    work_to: int = Field(..., ge=0, lt=_MAX_MIN)
+    entertainment_from: int = Field(..., ge=0, lt=_MAX_MIN)
+    entertainment_to: int = Field(..., ge=0, lt=_MAX_MIN)
+    maintenance_from: int = Field(480, ge=0, lt=_MAX_MIN, description="Maintenance start (default 480 = 8:00 UTC)")
+    maintenance_to: int = Field(540, ge=0, lt=_MAX_MIN, description="Maintenance end; (to-from) mod 1440 in [1,180]")
+
+    @model_validator(mode="after")
+    def _check_maintenance(self):
+        _validate_maintenance_window(self.maintenance_from, self.maintenance_to)
+        return self
 
 
 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)
+    work_from: Optional[int] = Field(None, ge=0, lt=_MAX_MIN)
+    work_to: Optional[int] = Field(None, ge=0, lt=_MAX_MIN)
+    entertainment_from: Optional[int] = Field(None, ge=0, lt=_MAX_MIN)
+    entertainment_to: Optional[int] = Field(None, ge=0, lt=_MAX_MIN)
+    maintenance_from: Optional[int] = Field(None, ge=0, lt=_MAX_MIN)
+    maintenance_to: Optional[int] = Field(None, ge=0, lt=_MAX_MIN)
+
+    @model_validator(mode="after")
+    def _check_maintenance(self):
+        # Only validate when both fields are present together; partial-
+        # update validation against the merged row happens at apply time.
+        if self.maintenance_from is not None and self.maintenance_to is not None:
+            _validate_maintenance_window(self.maintenance_from, self.maintenance_to)
+        return self
 
 
 class ScheduleTypeResponse(BaseModel):
@@ -27,6 +71,9 @@ class ScheduleTypeResponse(BaseModel):
     work_to: int
     entertainment_from: int
     entertainment_to: int
+    maintenance_from: int
+    maintenance_to: int
+    maintenance_duration_minutes: Optional[int] = None  # derived; populated by router
 
     class Config:
         from_attributes = True

app/schemas/schedule_type_special_slot.py

@@ -0,0 +1,43 @@
+"""Schemas for ScheduleTypeSpecialSlot CRUD (admin-only)."""
+
+from datetime import datetime
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+
+class SpecialSlotCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=64)
+    description: Optional[str] = Field(None, max_length=512)
+    minute_in_window: int = Field(0, ge=0, le=179, description="Minute offset (0-179) inside the schedule_type maintenance window")
+    estimated_duration: int = Field(15, ge=1, le=180, description="Duration in minutes; must fit inside the maintenance window (1-180min)")
+    priority: int = Field(50, ge=0, le=99)
+    event_data: Optional[dict[str, Any]] = Field(None, description="JSON payload merged into every materialised slot's event_data")
+    is_active: bool = True
+
+
+class SpecialSlotUpdate(BaseModel):
+    name: Optional[str] = Field(None, min_length=1, max_length=64)
+    description: Optional[str] = Field(None, max_length=512)
+    minute_in_window: Optional[int] = Field(None, ge=0, le=179)
+    estimated_duration: Optional[int] = Field(None, ge=1, le=180)
+    priority: Optional[int] = Field(None, ge=0, le=99)
+    event_data: Optional[dict[str, Any]] = None
+    is_active: Optional[bool] = None
+
+
+class SpecialSlotResponse(BaseModel):
+    id: int
+    schedule_type_id: int
+    name: str
+    description: Optional[str]
+    minute_in_window: int
+    estimated_duration: int
+    priority: int
+    event_data: Optional[dict[str, Any]]
+    is_active: bool
+    created_by_user_id: int
+    created_at: datetime
+
+    class Config:
+        from_attributes = True

app/schemas/schemas.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from typing import Optional, List
 from datetime import datetime, time
 from enum import Enum
@@ -186,6 +186,19 @@ class UserUpdate(BaseModel):
     discord_user_id: Optional[str] = None
 
 
+class UserBindAgentRequest(BaseModel):
+    """Request body for PATCH /users/{identifier}/bind-agent.
+
+    Binds an existing user to (agent_id, claw_identifier) by inserting a
+    row in the `agents` table. Both fields required (mirrors the
+    create-time invariant in UserCreate). Idempotent: re-binding the same
+    user to the same (agent_id, claw_identifier) returns the existing
+    Agent row instead of 409.
+    """
+    agent_id: str = Field(..., min_length=1, max_length=128)
+    claw_identifier: str = Field(..., min_length=1, max_length=128)
+
+
 class UserResponse(UserBase):
     id: int
     is_active: bool

app/services/special_slot_materialiser.py

@@ -0,0 +1,175 @@
+"""Materialise schedule_type special slots into per-agent time_slots rows.
+
+A ScheduleTypeSpecialSlot is a template — it lives on the schedule_type.
+For an agent on that schedule_type to actually be woken, the system must
+emit a row in `time_slots` with `slot_type=system`, `is_admin_locked=true`,
+`special_slot_id=<template_id>` for the agent's `user_id` on the target
+date. This module is the single materialisation point.
+
+Called from:
+  * GET /calendar/day — before returning slots, materialise today's special
+    slots for the calling user.
+  * GET /calendar/sync — before returning per-claw schedules, materialise
+    today's special slots for every agent on this claw whose schedule_type
+    has any active special slot template.
+
+Idempotent: re-running on the same (agent, date, special_slot_template)
+is a no-op — uniqueness is enforced via SELECT-then-insert. We do not add
+a DB-level unique constraint because the time_slots table is already
+indexed by (user_id, date) and an extra composite index is overkill for
+the low cardinality of (agents × special-slot-templates) per day.
+"""
+
+from __future__ import annotations
+
+from datetime import date as date_type, time as time_type
+from typing import Iterable
+
+from sqlalchemy.orm import Session
+
+from app.models.agent import Agent
+from app.models.calendar import TimeSlot, SlotType, SlotStatus, EventType
+from app.models.schedule_type import ScheduleType
+from app.models.schedule_type_special_slot import ScheduleTypeSpecialSlot
+
+
+def materialise_special_slots_for_user(
+    db: Session,
+    user_id: int,
+    target_date: date_type,
+    commit: bool = True,
+) -> list[TimeSlot]:
+    """Materialise today's special slots for one agent (identified by user_id).
+
+    Returns the list of newly created rows (may be empty if all already exist
+    or the agent has no schedule_type / no active templates).
+    """
+    agent = db.query(Agent).filter(Agent.user_id == user_id).first()
+    if not agent or not agent.schedule_type_id:
+        return []
+
+    return _materialise_for_agent(db, agent, target_date, commit=commit)
+
+
+def materialise_special_slots_for_claw(
+    db: Session,
+    claw_identifier: str,
+    target_date: date_type,
+    commit: bool = True,
+) -> list[TimeSlot]:
+    """Materialise today's special slots for every agent on a claw instance.
+
+    Used by the multi-agent `/calendar/sync` endpoint so plugin-driven
+    `runSync` cycles see the special slots without each agent having to
+    hit `/calendar/day` first.
+    """
+    agents = (
+        db.query(Agent)
+        .filter(
+            Agent.claw_identifier == claw_identifier,
+            Agent.schedule_type_id.isnot(None),
+        )
+        .all()
+    )
+    created: list[TimeSlot] = []
+    for agent in agents:
+        created.extend(_materialise_for_agent(db, agent, target_date, commit=False))
+    if commit and created:
+        db.commit()
+    return created
+
+
+def _materialise_for_agent(
+    db: Session,
+    agent: Agent,
+    target_date: date_type,
+    commit: bool,
+) -> list[TimeSlot]:
+    st: ScheduleType | None = (
+        db.query(ScheduleType).filter(ScheduleType.id == agent.schedule_type_id).first()
+    )
+    if not st:
+        return []
+
+    templates: Iterable[ScheduleTypeSpecialSlot] = (
+        db.query(ScheduleTypeSpecialSlot)
+        .filter(
+            ScheduleTypeSpecialSlot.schedule_type_id == st.id,
+            ScheduleTypeSpecialSlot.is_active.is_(True),
+        )
+        .all()
+    )
+
+    created: list[TimeSlot] = []
+    for tpl in templates:
+        if _already_materialised(db, agent.user_id, target_date, tpl.id):
+            continue
+        slot = _build_time_slot_from_template(
+            user_id=agent.user_id,
+            target_date=target_date,
+            schedule_type=st,
+            template=tpl,
+        )
+        db.add(slot)
+        created.append(slot)
+
+    if commit and created:
+        db.commit()
+        for slot in created:
+            db.refresh(slot)
+    return created
+
+
+def _already_materialised(
+    db: Session,
+    user_id: int,
+    target_date: date_type,
+    template_id: int,
+) -> bool:
+    return (
+        db.query(TimeSlot.id)
+        .filter(
+            TimeSlot.user_id == user_id,
+            TimeSlot.date == target_date,
+            TimeSlot.special_slot_id == template_id,
+        )
+        .first()
+        is not None
+    )
+
+
+def _build_time_slot_from_template(
+    *,
+    user_id: int,
+    target_date: date_type,
+    schedule_type: ScheduleType,
+    template: ScheduleTypeSpecialSlot,
+) -> TimeSlot:
+    # schedule_type.maintenance_from is minutes-since-UTC-midnight; the
+    # template's minute_in_window is an offset inside that window. Combined
+    # offset must fit in [0, 1440) and produce a wall-clock time_type.
+    total_min = (schedule_type.maintenance_from + template.minute_in_window) % 1440
+    scheduled_at = time_type(hour=total_min // 60, minute=total_min % 60, second=0)
+    # Merge admin-supplied event_data with bookkeeping pointers so the
+    # agent (and ARD) can identify the template at wake time.
+    merged_event_data = dict(template.event_data or {})
+    merged_event_data.setdefault("source", "schedule_type_special_slot")
+    merged_event_data["special_slot_id"] = template.id
+    merged_event_data["special_slot_name"] = template.name
+    merged_event_data["schedule_type_id"] = schedule_type.id
+    merged_event_data["schedule_type_name"] = schedule_type.name
+
+    return TimeSlot(
+        user_id=user_id,
+        date=target_date,
+        slot_type=SlotType.SYSTEM,
+        estimated_duration=template.estimated_duration,
+        scheduled_at=scheduled_at,
+        attended=False,
+        event_type=EventType.SYSTEM_EVENT,
+        event_data=merged_event_data,
+        priority=template.priority,
+        status=SlotStatus.NOT_STARTED,
+        is_admin_locked=True,
+        special_slot_id=template.id,
+    )