feat: add scheduler service

README.md

@@ -24,6 +24,7 @@
 - `knowledge-service`
 - `event-service`
 - `auth-service`
+- `scheduler-service`
 - `tool-service`
 
 每个服务都提供了最小 `FastAPI` 启动入口和健康检查接口，数据库相关服务也已经带上了 `SQLAlchemy` 模型骨架与 Alembic 目录。
@@ -61,6 +62,7 @@ pip install -e .\services\human-service
 pip install -e .\services\knowledge-service
 pip install -e .\services\event-service
 pip install -e .\services\auth-service
+pip install -e .\services\scheduler-service
 pip install -e .\services\tool-service
 ```
 
@@ -209,6 +211,7 @@ services/
   knowledge-service/
   event-service/
   auth-service/
+  scheduler-service/
   tool-service/
 libs/
   core-domain/
@@ -547,6 +550,41 @@ Invoke-RestMethod -Method Post `
 
 Through `api-gateway`, use `/gateway/auth/**`.
 
+## Scheduler Service APIs
+
+`scheduler-service` stores delayed jobs and due-job leases for time-based
+automation. It is intentionally service-neutral: jobs can target HTTP,
+event, runtime, agent, or team execution.
+
+Create a scheduled job:
+
+```powershell
+Invoke-RestMethod -Method Post `
+  -Uri http://127.0.0.1:8015/scheduler/jobs `
+  -ContentType "application/json" `
+  -Body '{"tenant_id":"t1","job_type":"runtime","name":"Run workflow later","schedule_time":"2026-04-26T12:00:00Z","payload_json":{"workflow_run_id":"run-id"}}'
+```
+
+Claim due jobs for a worker:
+
+```powershell
+Invoke-RestMethod -Method Post `
+  -Uri http://127.0.0.1:8015/scheduler/jobs/claim-due `
+  -ContentType "application/json" `
+  -Body '{"tenant_id":"t1","worker_key":"scheduler-worker-1","limit":20}'
+```
+
+Mark a job completed or failed:
+
+```powershell
+Invoke-RestMethod -Method Post `
+  -Uri http://127.0.0.1:8015/scheduler/jobs/job-id/status `
+  -ContentType "application/json" `
+  -Body '{"tenant_id":"t1","status":"completed"}'
+```
+
+Through `api-gateway`, use `/gateway/scheduler/**`.
+
 Execute an agent run without calling an external model:
 
 ```powershell
@@ -923,6 +961,7 @@ $env:AGENT_PLATFORM_SMOKE_RUNTIME_URL="http://127.0.0.1:8000/gateway/runtime"
 - `/gateway/knowledge/**` -> `knowledge-service /knowledge/**`
 - `/gateway/events/**` -> `event-service /events/**`
 - `/gateway/auth/**` -> `auth-service /auth/**`
+- `/gateway/scheduler/**` -> `scheduler-service /scheduler/**`
 - `/gateway/tools/**` -> `tool-service /tools/**`
 - `/gateway/models/**` -> `model-gateway-service /models/**`
 - `/gateway/code/**` -> `code-runner-service /code/**`
@@ -1169,6 +1208,7 @@ Important notes:
 - `knowledge-service` stores knowledge bases, documents, chunks, and local retrieval metadata under `/data`
 - `event-service` stores platform events and delivery status under `/data`
 - `auth-service` stores users, roles, assignments, and permission policy metadata under `/data`
+- `scheduler-service` stores delayed jobs, due-job leases, and retry status under `/data`
 - `agent-worker` has no exposed port and can be scaled independently; set `AGENT_PLATFORM_AGENT_WORKER_DRY_RUN=true` for no-key local smoke runs
 - `runtime-worker` has no exposed port and can be scaled independently; prefer PostgreSQL for real multi-worker write concurrency
 - `runtime-service` automatically resolves internal URLs to `workflow-service`, `tool-service`, `model-gateway-service`, and `code-runner-service`

deployments/docker/docker-compose.yml

@@ -333,6 +333,26 @@ services:
       timeout: 5s
       retries: 5
 
+  scheduler-service:
+    build:
+      context: ../..
+      dockerfile: deployments/docker/python-service.Dockerfile
+      args:
+        SERVICE_PATH: services/scheduler-service
+    container_name: agent-platform-scheduler-service
+    command: ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8015"]
+    environment:
+      AGENT_PLATFORM_DATABASE_URL: sqlite:////data/scheduler_service.db
+    ports:
+      - "8015:8015"
+    volumes:
+      - scheduler_service_data:/data
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8015/scheduler/health').read()"]
+      interval: 15s
+      timeout: 5s
+      retries: 5
+
   runtime-service:
     build:
       context: ../..
@@ -354,6 +374,7 @@ services:
       AGENT_PLATFORM_HUMAN_SERVICE_URL: http://human-service:8011
       AGENT_PLATFORM_KNOWLEDGE_SERVICE_URL: http://knowledge-service:8012
       AGENT_PLATFORM_EVENT_SERVICE_URL: http://event-service:8013
+      AGENT_PLATFORM_SCHEDULER_SERVICE_URL: http://scheduler-service:8015
     ports:
       - "8003:8003"
     volumes:
@@ -381,6 +402,8 @@ services:
         condition: service_started
       event-service:
         condition: service_started
+      scheduler-service:
+        condition: service_started
     healthcheck:
       test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8003/runtime/health').read()"]
       interval: 15s
@@ -402,6 +425,7 @@ services:
       AGENT_PLATFORM_CODE_RUNNER_SERVICE_URL: http://code-runner-service:8006
       AGENT_PLATFORM_KNOWLEDGE_SERVICE_URL: http://knowledge-service:8012
       AGENT_PLATFORM_EVENT_SERVICE_URL: http://event-service:8013
+      AGENT_PLATFORM_SCHEDULER_SERVICE_URL: http://scheduler-service:8015
       AGENT_PLATFORM_WORKER_POLL_INTERVAL_SECONDS: ${AGENT_PLATFORM_WORKER_POLL_INTERVAL_SECONDS:-1}
       AGENT_PLATFORM_WORKER_LEASE_SECONDS: ${AGENT_PLATFORM_WORKER_LEASE_SECONDS:-300}
     volumes:
@@ -419,6 +443,8 @@ services:
         condition: service_started
       event-service:
         condition: service_started
+      scheduler-service:
+        condition: service_started
 
   api-gateway:
     build:
@@ -444,6 +470,7 @@ services:
       AGENT_PLATFORM_KNOWLEDGE_SERVICE_URL: http://knowledge-service:8012
       AGENT_PLATFORM_EVENT_SERVICE_URL: http://event-service:8013
       AGENT_PLATFORM_AUTH_SERVICE_URL: http://auth-service:8014
+      AGENT_PLATFORM_SCHEDULER_SERVICE_URL: http://scheduler-service:8015
       AGENT_PLATFORM_AUTH_REQUIRED: ${AGENT_PLATFORM_AUTH_REQUIRED:-false}
     ports:
       - "8000:8000"
@@ -478,6 +505,8 @@ services:
         condition: service_started
       auth-service:
         condition: service_started
+      scheduler-service:
+        condition: service_started
     healthcheck:
       test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health').read()"]
       interval: 15s
@@ -494,6 +523,7 @@ volumes:
   knowledge_service_data:
   event_service_data:
   auth_service_data:
+  scheduler_service_data:
   workflow_service_data:
   session_service_data:
   runtime_service_data:

libs/core-domain/src/core_domain/__init__.py

@@ -66,6 +66,11 @@ from .runtime_contracts import (
     WorkflowRunStatusUpdateContract,
     WorkflowRunContract,
 )
+from .scheduler_contracts import (
+    ScheduledJobContract,
+    ScheduledJobStatus,
+    ScheduledJobType,
+)
 from .service import ServiceDescriptor, ServiceHealth
 from .skill_contracts import (
     SkillDefinitionContract,
@@ -146,6 +151,9 @@ __all__ = [
     "NodeRunStatusUpdateContract",
     "RunBootstrapContract",
     "RunCreateContract",
+    "ScheduledJobContract",
+    "ScheduledJobStatus",
+    "ScheduledJobType",
     "ServiceDescriptor",
     "ServiceHealth",
     "SkillDefinitionContract",

libs/core-domain/src/core_domain/scheduler_contracts.py

@@ -0,0 +1,39 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from core_shared import JSONValue
+
+
+ScheduledJobStatus = Literal[
+    "scheduled",
+    "claimed",
+    "completed",
+    "failed",
+    "cancelled",
+]
+ScheduledJobType = Literal["http", "event", "runtime", "agent", "team"]
+
+
+class ScheduledJobContract(BaseModel):
+    id: str
+    tenant_id: str
+    job_type: ScheduledJobType
+    status: ScheduledJobStatus
+    name: str
+    description: str | None = None
+    target_service: str | None = None
+    target_url: str | None = None
+    method: str | None = None
+    payload_json: dict[str, JSONValue] = Field(default_factory=dict)
+    schedule_time: datetime
+    lease_expire_time: datetime | None = None
+    claimed_by: str | None = None
+    claimed_time: datetime | None = None
+    completed_time: datetime | None = None
+    attempt_count: int = 0
+    max_attempts: int = 3
+    last_error_message: str | None = None
+    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)
+    created_time: datetime

services/api-gateway/app/api/routes.py

@@ -199,6 +199,12 @@ def build_proxy_targets(settings: ApiGatewaySettings) -> dict[ProxyServiceName,
             path_prefix="/auth",
             health_path="/auth/health",
         ),
+        "scheduler-service": ProxyTarget(
+            service_name="scheduler-service",
+            base_url=settings.scheduler_service_url,
+            path_prefix="/scheduler",
+            health_path="/scheduler/health",
+        ),
     }
 
 
@@ -449,6 +455,27 @@ async def proxy_auth_service(
     )
 
 
+@router.api_route(
+    "/gateway/scheduler",
+    methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
+)
+@router.api_route(
+    "/gateway/scheduler/{path:path}",
+    methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
+)
+async def proxy_scheduler_service(
+    request: Request,
+    path: str = "",
+    settings: ApiGatewaySettings = Depends(get_gateway_settings),
+    proxy: ServiceProxy = Depends(get_service_proxy),
+) -> Response:
+    return await proxy.forward(
+        request=request,
+        target=build_proxy_targets(settings)["scheduler-service"],
+        path=path,
+    )
+
+
 @router.api_route(
     "/gateway/tools",
     methods=["GET", "POST", "PUT", "PATCH", "DELETE"],

services/api-gateway/app/bootstrap/settings.py

@@ -19,6 +19,7 @@ class ApiGatewaySettings(ServiceSettings):
     knowledge_service_url: str = "http://127.0.0.1:8012"
     event_service_url: str = "http://127.0.0.1:8013"
     auth_service_url: str = "http://127.0.0.1:8014"
+    scheduler_service_url: str = "http://127.0.0.1:8015"
     proxy_timeout_seconds: float = 30.0
     downstream_health_timeout_seconds: float = 2.0
     auth_required: bool = False

services/api-gateway/app/infrastructure/proxy.py

@@ -27,6 +27,7 @@ ProxyServiceName = Literal[
     "knowledge-service",
     "event-service",
     "auth-service",
+    "scheduler-service",
 ]
 
 

services/scheduler-service/alembic.ini

@@ -0,0 +1,37 @@
+[alembic]
+script_location = alembic
+prepend_sys_path = .
+sqlalchemy.url = sqlite:///./scheduler_service.db
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

services/scheduler-service/alembic/env.py

@@ -0,0 +1,38 @@
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+
+from app.db.models import Base
+
+config = context.config
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def run_migrations_offline() -> None:
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

services/scheduler-service/alembic/versions/.gitkeep

@@ -0,0 +1 @@
+

services/scheduler-service/alembic/versions/20260426_0001_init_scheduler_models.py

@@ -0,0 +1,65 @@
+"""init scheduler models
+
+Revision ID: 20260426_0001
+Revises:
+Create Date: 2026-04-26 00:20:00
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "20260426_0001"
+down_revision: str | None = None
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "scheduled_job",
+        sa.Column("job_type", sa.String(length=32), nullable=False),
+        sa.Column("status", sa.String(length=32), nullable=False),
+        sa.Column("name", sa.String(length=128), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("target_service", sa.String(length=64), nullable=True),
+        sa.Column("target_url", sa.Text(), nullable=True),
+        sa.Column("method", sa.String(length=16), nullable=True),
+        sa.Column("payload_json", sa.JSON(), nullable=False),
+        sa.Column("schedule_time", sa.DateTime(), nullable=False),
+        sa.Column("lease_expire_time", sa.DateTime(), nullable=True),
+        sa.Column("claimed_by", sa.String(length=128), nullable=True),
+        sa.Column("claimed_time", sa.DateTime(), nullable=True),
+        sa.Column("completed_time", sa.DateTime(), nullable=True),
+        sa.Column("attempt_count", sa.Integer(), nullable=False),
+        sa.Column("max_attempts", sa.Integer(), nullable=False),
+        sa.Column("last_error_message", sa.Text(), nullable=True),
+        sa.Column("metadata_json", sa.JSON(), nullable=False),
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("created_by", sa.String(length=36), nullable=True),
+        sa.Column("updated_by", sa.String(length=36), nullable=True),
+        sa.Column("created_time", sa.DateTime(), nullable=False),
+        sa.Column("updated_time", sa.DateTime(), nullable=False),
+        sa.Column("deleted_time", sa.DateTime(), nullable=True),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_scheduled_job_tenant_id", "scheduled_job", ["tenant_id"])
+    op.create_index("ix_scheduled_job_job_type", "scheduled_job", ["job_type"])
+    op.create_index("ix_scheduled_job_status", "scheduled_job", ["status"])
+    op.create_index("ix_scheduled_job_target_service", "scheduled_job", ["target_service"])
+    op.create_index("ix_scheduled_job_schedule_time", "scheduled_job", ["schedule_time"])
+    op.create_index("ix_scheduled_job_claimed_by", "scheduled_job", ["claimed_by"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_scheduled_job_claimed_by", table_name="scheduled_job")
+    op.drop_index("ix_scheduled_job_schedule_time", table_name="scheduled_job")
+    op.drop_index("ix_scheduled_job_target_service", table_name="scheduled_job")
+    op.drop_index("ix_scheduled_job_status", table_name="scheduled_job")
+    op.drop_index("ix_scheduled_job_job_type", table_name="scheduled_job")
+    op.drop_index("ix_scheduled_job_tenant_id", table_name="scheduled_job")
+    op.drop_table("scheduled_job")

services/scheduler-service/app/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/api/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/api/routes.py

@@ -0,0 +1,79 @@
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from core_domain import ScheduledJobStatus, ScheduledJobType, ServiceHealth
+
+from app.application.services import SchedulerApplicationService
+from app.db.session import get_db
+from app.domain.repositories import ScheduledJobRepository
+from app.schemas.scheduler import (
+    DueJobClaimRequest,
+    ScheduledJobCreateRequest,
+    ScheduledJobResponse,
+    ScheduledJobStatusUpdateRequest,
+)
+
+router = APIRouter()
+
+
+def get_scheduler_application_service(
+    db: Session = Depends(get_db),
+) -> SchedulerApplicationService:
+    return SchedulerApplicationService(job_repository=ScheduledJobRepository(db))
+
+
+@router.get("/health", response_model=ServiceHealth)
+def health_check(db: Session = Depends(get_db)) -> ServiceHealth:
+    db.execute(text("SELECT 1"))
+    return ServiceHealth(service="scheduler-service", status="ok", database="ok")
+
+
+@router.post("/jobs", response_model=ScheduledJobResponse)
+def create_job(
+    payload: ScheduledJobCreateRequest,
+    service: SchedulerApplicationService = Depends(get_scheduler_application_service),
+) -> ScheduledJobResponse:
+    return ScheduledJobResponse.from_entity(service.create_job(payload))
+
+
+@router.get("/jobs", response_model=list[ScheduledJobResponse])
+def list_jobs(
+    tenant_id: str = Query(...),
+    status: ScheduledJobStatus | None = Query(default=None),
+    job_type: ScheduledJobType | None = Query(default=None),
+    limit: int = Query(default=100, ge=1, le=500),
+    service: SchedulerApplicationService = Depends(get_scheduler_application_service),
+) -> list[ScheduledJobResponse]:
+    return [
+        ScheduledJobResponse.from_entity(item)
+        for item in service.list_jobs(
+            tenant_id=tenant_id,
+            status=status,
+            job_type=job_type,
+            limit=limit,
+        )
+    ]
+
+
+@router.post("/jobs/claim-due", response_model=list[ScheduledJobResponse])
+def claim_due_jobs(
+    payload: DueJobClaimRequest,
+    service: SchedulerApplicationService = Depends(get_scheduler_application_service),
+) -> list[ScheduledJobResponse]:
+    return [
+        ScheduledJobResponse.from_entity(item)
+        for item in service.claim_due_jobs(payload)
+    ]
+
+
+@router.post("/jobs/{job_id}/status", response_model=ScheduledJobResponse)
+def update_job_status(
+    job_id: str,
+    payload: ScheduledJobStatusUpdateRequest,
+    service: SchedulerApplicationService = Depends(get_scheduler_application_service),
+) -> ScheduledJobResponse:
+    entity = service.update_job_status(job_id=job_id, payload=payload)
+    if entity is None:
+        raise HTTPException(status_code=404, detail=f"scheduled job not found: {job_id}")
+    return ScheduledJobResponse.from_entity(entity)

services/scheduler-service/app/application/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/application/services.py

@@ -0,0 +1,70 @@
+from datetime import datetime
+
+from core_domain import ScheduledJobStatus, ScheduledJobType
+
+from app.db.models import ScheduledJob
+from app.domain.repositories import ScheduledJobRepository
+from app.schemas.scheduler import (
+    DueJobClaimRequest,
+    ScheduledJobCreateRequest,
+    ScheduledJobStatusUpdateRequest,
+)
+
+
+class SchedulerApplicationService:
+    def __init__(self, *, job_repository: ScheduledJobRepository) -> None:
+        self.job_repository = job_repository
+
+    def create_job(self, payload: ScheduledJobCreateRequest) -> ScheduledJob:
+        return self.job_repository.create(
+            tenant_id=payload.tenant_id,
+            job_type=payload.job_type,
+            name=payload.name,
+            description=payload.description,
+            target_service=payload.target_service,
+            target_url=payload.target_url,
+            method=payload.method,
+            payload_json=payload.payload_json,
+            schedule_time=payload.schedule_time,
+            max_attempts=payload.max_attempts,
+            metadata_json=payload.metadata_json,
+        )
+
+    def list_jobs(
+        self,
+        *,
+        tenant_id: str,
+        status: ScheduledJobStatus | None,
+        job_type: ScheduledJobType | None,
+        limit: int,
+    ) -> list[ScheduledJob]:
+        return self.job_repository.list_by_scope(
+            tenant_id=tenant_id,
+            status=status,
+            job_type=job_type,
+            limit=limit,
+        )
+
+    def claim_due_jobs(self, payload: DueJobClaimRequest) -> list[ScheduledJob]:
+        return self.job_repository.claim_due_jobs(
+            tenant_id=payload.tenant_id,
+            worker_key=payload.worker_key,
+            lease_seconds=payload.lease_seconds,
+            limit=payload.limit,
+            now_time=payload.now_time or datetime.utcnow(),
+        )
+
+    def update_job_status(
+        self,
+        *,
+        job_id: str,
+        payload: ScheduledJobStatusUpdateRequest,
+    ) -> ScheduledJob | None:
+        entity = self.job_repository.get_by_id(tenant_id=payload.tenant_id, job_id=job_id)
+        if entity is None:
+            return None
+        return self.job_repository.update_status(
+            job_id=job_id,
+            status=payload.status,
+            last_error_message=payload.last_error_message,
+        )

services/scheduler-service/app/bootstrap/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/bootstrap/app.py

@@ -0,0 +1,17 @@
+from fastapi import FastAPI
+
+from app.api.routes import router
+from app.bootstrap.settings import SchedulerServiceSettings
+from app.db.session import build_session_factory
+
+
+def create_app() -> FastAPI:
+    settings = SchedulerServiceSettings()
+    app = FastAPI(
+        title="agent-platform scheduler-service",
+        version="0.1.0",
+    )
+    app.state.settings = settings
+    app.state.session_factory = build_session_factory(settings)
+    app.include_router(router, prefix="/scheduler", tags=["scheduler"])
+    return app

services/scheduler-service/app/bootstrap/settings.py

@@ -0,0 +1,9 @@
+from core_shared import ServiceSettings
+
+
+class SchedulerServiceSettings(ServiceSettings):
+    service_name: str = "scheduler-service"
+    service_port: int = 8015
+    database_url: str = "sqlite:///./scheduler_service.db"
+    default_lease_seconds: int = 300
+    default_claim_limit: int = 50

services/scheduler-service/app/db/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/db/models/__init__.py

@@ -0,0 +1,5 @@
+from core_db import Base
+
+from .scheduled_job import ScheduledJob
+
+__all__ = ["Base", "ScheduledJob"]

services/scheduler-service/app/db/models/scheduled_job.py

@@ -0,0 +1,30 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer, String, Text
+from sqlalchemy.dialects.sqlite import JSON
+from sqlalchemy.orm import Mapped, mapped_column
+
+from core_db import AuditMixin, Base, TenantMixin, VersionMixin
+from core_shared import JSONValue
+
+
+class ScheduledJob(TenantMixin, AuditMixin, VersionMixin, Base):
+    __tablename__ = "scheduled_job"
+
+    job_type: Mapped[str] = mapped_column(String(32), index=True)
+    status: Mapped[str] = mapped_column(String(32), default="scheduled", index=True)
+    name: Mapped[str] = mapped_column(String(128))
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    target_service: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    target_url: Mapped[str | None] = mapped_column(Text, nullable=True)
+    method: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    payload_json: Mapped[dict[str, JSONValue]] = mapped_column(JSON, default=dict)
+    schedule_time: Mapped[datetime] = mapped_column(DateTime, index=True)
+    lease_expire_time: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    claimed_by: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    claimed_time: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    completed_time: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    attempt_count: Mapped[int] = mapped_column(Integer, default=0)
+    max_attempts: Mapped[int] = mapped_column(Integer, default=3)
+    last_error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    metadata_json: Mapped[dict[str, JSONValue]] = mapped_column(JSON, default=dict)

services/scheduler-service/app/db/session.py

@@ -0,0 +1,30 @@
+from collections.abc import Generator
+
+from fastapi import Request
+from sqlalchemy.orm import Session, sessionmaker
+
+from core_db import DatabaseSettings, create_engine_from_settings, create_session_factory
+
+from app.bootstrap.settings import SchedulerServiceSettings
+
+
+def build_session_factory(
+    settings: SchedulerServiceSettings | None = None,
+) -> sessionmaker[Session]:
+    resolved_settings = settings or SchedulerServiceSettings()
+    engine = create_engine_from_settings(
+        DatabaseSettings(
+            database_url=resolved_settings.database_url,
+            echo_sql=resolved_settings.echo_sql,
+        )
+    )
+    return create_session_factory(engine)
+
+
+def get_db(request: Request) -> Generator[Session, None, None]:
+    session_factory: sessionmaker[Session] = request.app.state.session_factory
+    session = session_factory()
+    try:
+        yield session
+    finally:
+        session.close()

services/scheduler-service/app/domain/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/domain/repositories.py

@@ -0,0 +1,147 @@
+from datetime import datetime, timedelta
+
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from core_domain import ScheduledJobStatus, ScheduledJobType
+from core_shared import JSONValue
+
+from app.db.models import ScheduledJob
+
+
+class ScheduledJobRepository:
+    def __init__(self, db: Session) -> None:
+        self.db = db
+
+    def create(
+        self,
+        *,
+        tenant_id: str,
+        job_type: ScheduledJobType,
+        name: str,
+        description: str | None,
+        target_service: str | None,
+        target_url: str | None,
+        method: str | None,
+        payload_json: dict[str, JSONValue],
+        schedule_time: datetime,
+        max_attempts: int,
+        metadata_json: dict[str, JSONValue],
+    ) -> ScheduledJob:
+        entity = ScheduledJob(
+            tenant_id=tenant_id,
+            job_type=job_type,
+            name=name,
+            description=description,
+            target_service=target_service,
+            target_url=target_url,
+            method=method,
+            payload_json=payload_json,
+            schedule_time=schedule_time,
+            max_attempts=max_attempts,
+            metadata_json=metadata_json,
+        )
+        self.db.add(entity)
+        self.db.commit()
+        self.db.refresh(entity)
+        return entity
+
+    def list_by_scope(
+        self,
+        *,
+        tenant_id: str,
+        status: ScheduledJobStatus | None = None,
+        job_type: ScheduledJobType | None = None,
+        limit: int = 100,
+    ) -> list[ScheduledJob]:
+        stmt = select(ScheduledJob).where(ScheduledJob.tenant_id == tenant_id)
+        if status is not None:
+            stmt = stmt.where(ScheduledJob.status == status)
+        if job_type is not None:
+            stmt = stmt.where(ScheduledJob.job_type == job_type)
+        stmt = stmt.order_by(ScheduledJob.schedule_time.asc()).limit(limit)
+        return list(self.db.scalars(stmt))
+
+    def get_by_id(self, *, tenant_id: str, job_id: str) -> ScheduledJob | None:
+        stmt = (
+            select(ScheduledJob)
+            .where(ScheduledJob.tenant_id == tenant_id)
+            .where(ScheduledJob.id == job_id)
+        )
+        return self.db.scalar(stmt)
+
+    def claim_due_jobs(
+        self,
+        *,
+        tenant_id: str,
+        worker_key: str,
+        lease_seconds: int,
+        limit: int,
+        now_time: datetime,
+    ) -> list[ScheduledJob]:
+        self.release_expired_leases(now_time=now_time)
+        stmt = (
+            select(ScheduledJob)
+            .where(ScheduledJob.tenant_id == tenant_id)
+            .where(ScheduledJob.status == "scheduled")
+            .where(ScheduledJob.schedule_time <= now_time)
+            .where(ScheduledJob.attempt_count < ScheduledJob.max_attempts)
+            .order_by(ScheduledJob.schedule_time.asc())
+            .limit(limit)
+        )
+        jobs = list(self.db.scalars(stmt))
+        lease_expire_time = now_time + timedelta(seconds=lease_seconds)
+        for job in jobs:
+            job.status = "claimed"
+            job.claimed_by = worker_key
+            job.claimed_time = now_time
+            job.lease_expire_time = lease_expire_time
+            job.attempt_count += 1
+        if jobs:
+            self.db.commit()
+            for job in jobs:
+                self.db.refresh(job)
+        return jobs
+
+    def release_expired_leases(self, *, now_time: datetime, max_items: int = 100) -> int:
+        stmt = (
+            select(ScheduledJob)
+            .where(ScheduledJob.status == "claimed")
+            .where(
+                or_(
+                    ScheduledJob.lease_expire_time.is_(None),
+                    ScheduledJob.lease_expire_time <= now_time,
+                )
+            )
+            .limit(max_items)
+        )
+        jobs = list(self.db.scalars(stmt))
+        for job in jobs:
+            job.status = "scheduled" if job.attempt_count < job.max_attempts else "failed"
+            job.claimed_by = None
+            job.claimed_time = None
+            job.lease_expire_time = None
+            if job.status == "failed":
+                job.last_error_message = "lease expired and max attempts reached"
+        if jobs:
+            self.db.commit()
+        return len(jobs)
+
+    def update_status(
+        self,
+        *,
+        job_id: str,
+        status: ScheduledJobStatus,
+        last_error_message: str | None = None,
+    ) -> ScheduledJob | None:
+        entity = self.db.get(ScheduledJob, job_id)
+        if entity is None:
+            return None
+        entity.status = status
+        entity.last_error_message = last_error_message
+        if status in {"completed", "failed", "cancelled"}:
+            entity.completed_time = datetime.utcnow()
+            entity.lease_expire_time = None
+        self.db.commit()
+        self.db.refresh(entity)
+        return entity

services/scheduler-service/app/main.py

@@ -0,0 +1,3 @@
+from app.bootstrap.app import create_app
+
+app = create_app()

services/scheduler-service/app/schemas/__init__.py

@@ -0,0 +1 @@
+

services/scheduler-service/app/schemas/scheduler.py

@@ -0,0 +1,44 @@
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel, Field
+
+from core_domain import ScheduledJobContract, ScheduledJobStatus, ScheduledJobType
+from core_shared import JSONValue
+
+if TYPE_CHECKING:
+    from app.db.models import ScheduledJob
+
+
+class ScheduledJobCreateRequest(BaseModel):
+    tenant_id: str
+    job_type: ScheduledJobType
+    name: str
+    description: str | None = None
+    target_service: str | None = None
+    target_url: str | None = None
+    method: str | None = None
+    payload_json: dict[str, JSONValue] = Field(default_factory=dict)
+    schedule_time: datetime
+    max_attempts: int = Field(default=3, ge=1, le=20)
+    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)
+
+
+class ScheduledJobResponse(ScheduledJobContract):
+    @classmethod
+    def from_entity(cls, entity: "ScheduledJob") -> "ScheduledJobResponse":
+        return cls.model_validate(entity, from_attributes=True)
+
+
+class ScheduledJobStatusUpdateRequest(BaseModel):
+    tenant_id: str
+    status: ScheduledJobStatus
+    last_error_message: str | None = None
+
+
+class DueJobClaimRequest(BaseModel):
+    tenant_id: str
+    worker_key: str
+    lease_seconds: int = Field(default=300, ge=1, le=86400)
+    limit: int = Field(default=50, ge=1, le=500)
+    now_time: datetime | None = None

services/scheduler-service/pyproject.toml

@@ -0,0 +1,25 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "scheduler-service"
+version = "0.1.0"
+description = "Scheduled job service for agent platform."
+requires-python = ">=3.11"
+dependencies = [
+  "alembic>=1.13,<2.0",
+  "fastapi>=0.111,<1.0",
+  "uvicorn[standard]>=0.30,<1.0",
+  "pydantic>=2.7,<3.0",
+  "sqlalchemy>=2.0,<3.0",
+  "core-db",
+  "core-domain",
+  "core-shared",
+]
+
+[tool.setuptools]
+package-dir = {"" = "."}
+
+[tool.setuptools.packages.find]
+where = ["."]