diff --git a/dashboard/src/api/types/cronjob.ts b/dashboard/src/api/types/cronjob.ts index e08f4f96..b701138e 100644 --- a/dashboard/src/api/types/cronjob.ts +++ b/dashboard/src/api/types/cronjob.ts @@ -67,6 +67,7 @@ export type CronJobViewLegacy = Record; */ export interface OctopCronRow { id: string; + name: string; agent_id: string; trigger: string; prompt: string; @@ -83,10 +84,12 @@ export interface OctopCronRow { /** Body sent to POST /api/agents/:id/cron */ export interface OctopCronCreateBody { + name?: string | null; trigger: string; prompt: string; session_key?: string | null; fresh_thread?: boolean; + enabled?: boolean; model?: string | null; task_type?: "text" | "agent"; mcp_servers?: string[]; @@ -94,6 +97,7 @@ export interface OctopCronCreateBody { /** Body sent to PATCH /api/agents/:id/cron/:cron_id */ export interface OctopCronPatchBody { + name?: string | null; trigger?: string; prompt?: string; session_key?: string | null; diff --git a/dashboard/src/locales/en.json b/dashboard/src/locales/en.json index 66ba02c2..802b3e18 100644 --- a/dashboard/src/locales/en.json +++ b/dashboard/src/locales/en.json @@ -1658,6 +1658,7 @@ "idTooltip": "A unique identifier for this job, auto-generated by the system.", "name": "Job Name", "nameTooltip": "A descriptive name to easily identify this job in the list.", + "nameTooLong": "Job name must be at most {{max}} characters", "enabled": "Enable Job", "enabledTooltip": "When enabled, the job runs automatically on schedule. Disable to pause execution.", "sectionSchedule": "Execution Frequency", diff --git a/dashboard/src/locales/zh.json b/dashboard/src/locales/zh.json index 886984d0..e61fa454 100644 --- a/dashboard/src/locales/zh.json +++ b/dashboard/src/locales/zh.json @@ -1656,6 +1656,7 @@ "idTooltip": "任务的唯一标识符，由系统自动生成。", "name": "任务名称", "nameTooltip": "给任务起一个易于辨识的名称，方便在列表中快速找到。", + "nameTooLong": "任务名称不能超过 {{max}} 个字符", "enabled": "启用任务", "enabledTooltip": "开启后任务将按照设定的时间自动执行，关闭则暂停执行。", "sectionSchedule": "执行频率", diff --git a/dashboard/src/pages/Control/CronJobs/components/JobDetailDrawer.tsx b/dashboard/src/pages/Control/CronJobs/components/JobDetailDrawer.tsx index cb721d6d..737f79f4 100644 --- a/dashboard/src/pages/Control/CronJobs/components/JobDetailDrawer.tsx +++ b/dashboard/src/pages/Control/CronJobs/components/JobDetailDrawer.tsx @@ -111,6 +111,9 @@ export function JobDetailDrawer({ {job.id} + + {job.name || "—"} + {job.enabled ? ( {t("common.enabled")} diff --git a/dashboard/src/pages/Control/CronJobs/components/JobDrawer.tsx b/dashboard/src/pages/Control/CronJobs/components/JobDrawer.tsx index 4a03a626..a15f4c5e 100644 --- a/dashboard/src/pages/Control/CronJobs/components/JobDrawer.tsx +++ b/dashboard/src/pages/Control/CronJobs/components/JobDrawer.tsx @@ -20,7 +20,7 @@ import { cronToPreset, presetToCron, } from "./constants"; -import { CRON_PROMPT_MAX_LEN } from "../constants"; +import { CRON_NAME_MAX_LEN, CRON_PROMPT_MAX_LEN } from "../constants"; import { channelFromSessionKey } from "../cronDisplay"; import type { CronJobFormValues } from "../useCronJobs"; import { @@ -220,6 +220,27 @@ export function JobDrawer({ )} + + + + ({ style: { paddingLeft: 28 } }), @@ -111,12 +111,33 @@ export const createColumns = ( ); }, }, + { + title: handlers.t("cronJobs.col.name"), + dataIndex: "name", + key: "name", + width: 180, + fixed: sticky ? "left" : undefined, + ellipsis: true, + render: (name: string, record: CronJob) => { + const text = name?.trim() || record.id; + return ( + + + + ); + }, + }, { title: handlers.t("common.enabled"), dataIndex: "enabled", key: "enabled", width: 100, - fixed: sticky ? "left" : undefined, render: (enabled: boolean) => ( { const taskType = record.task_type === "text" ? "text" : "agent"; return ( @@ -169,7 +191,7 @@ export const createColumns = ( { title: handlers.t("cronJobs.col.prompt"), key: "prompt", - width: 200, + width: 240, ellipsis: true, render: (_: unknown, record: CronJob) => { const content = extractPromptFromJob(record); @@ -252,8 +274,9 @@ export const createColumns = ( { title: handlers.t("cronJobs.action"), key: "action", - width: 200, + width: 220, fixed: sticky ? "right" : undefined, + className: styles.actionCell, render: (_: unknown, record: CronJob) => { const menuItems: MenuProps["items"] = [ { @@ -272,7 +295,7 @@ export const createColumns = ( ]; return ( -
+
) : ( - j.id === jobId); const optimistic = { ...original, + name: (values.name || "").trim(), enabled: values.enabled, task_type: values.task_type, model: values.model ?? undefined, diff --git a/docs/api.md b/docs/api.md index 2afcbcad..edf3d6d0 100644 --- a/docs/api.md +++ b/docs/api.md @@ -143,19 +143,20 @@ because each request is a one-shot continuation. | `GET` | `/settings/timezone` | user | process-level `{timezone}` from `default_timezone` | | `GET` | `/settings/upload` | user | `{max_upload_mb, max_upload_bytes}` from `max_upload_mb` | | `GET` | `/cron/settings` | user | compat alias of `/settings/timezone` | -| `GET` | `/agents/{aid}/cron` | owner | list of cron rows | -| `POST` | `/agents/{aid}/cron` | owner | body `{trigger, prompt, session_key?, fresh_thread?, model?, task_type?}` → `201` | -| `GET` | `/agents/{aid}/cron/{cid}` | owner | cron row | -| `PATCH` | `/agents/{aid}/cron/{cid}` | owner | body subset → updated row | -| `DELETE` | `/agents/{aid}/cron/{cid}` | owner | `204` | -| `POST` | `/agents/{aid}/cron/{cid}/run-now` | owner | `204` (fire immediately, off-schedule) | +| `GET` | `/agents/{aid}/cron` | agent access | list cron rows; shared-agent viewers see only their own rows | +| `POST` | `/agents/{aid}/cron` | agent access | body `{name?, trigger, prompt, session_key?, fresh_thread?, enabled?, model?, task_type?}` → `201` | +| `GET` | `/agents/{aid}/cron/{cid}` | owner/creator | cron row | +| `PATCH` | `/agents/{aid}/cron/{cid}` | owner/creator | body subset → updated row | +| `DELETE` | `/agents/{aid}/cron/{cid}` | owner/creator | `204` | +| `POST` | `/agents/{aid}/cron/{cid}/run-now` | owner/creator | `204` (fire immediately, off-schedule) | `task_type` is `"text"` (push prompt directly to the session) or `"agent"` (run the prompt through the LLM and push the reply). Default: `"agent"`. `trigger` accepts cron expressions (`"0 9 * * *"`) plus the `interval:N` / `date:ISO8601` aliases documented in `infra/cron/trigger.py`. `prompt` must be non-empty and -≤ 2000 characters. +≤ 2000 characters. `name` is an optional display label; when omitted, +the server derives one from `prompt`. ## Providers diff --git a/src/octop/api/routers/cron.py b/src/octop/api/routers/cron.py index 3f78e710..094dbf55 100644 --- a/src/octop/api/routers/cron.py +++ b/src/octop/api/routers/cron.py @@ -7,9 +7,13 @@ from fastapi import APIRouter, Depends from pydantic import BaseModel, Field -from octop.api.common.agent import require_agent_owner_row +from octop.api.common.agent import require_agent_row, user_owns_agent from octop.api.deps import current_user, get_server -from octop.infra.cron.task_type import normalize_cron_task_type, require_cron_prompt +from octop.infra.cron.task_type import ( + normalize_cron_task_type, + require_cron_name, + require_cron_prompt, +) from octop.infra.cron.trigger import build_trigger from octop.infra.errors import ErrorCode, OctopError from octop.infra.utils.ulid import new_cron_id @@ -18,16 +22,19 @@ class CronCreateBody(BaseModel): + name: str | None = None trigger: str prompt: str session_key: str | None = None fresh_thread: bool = False + enabled: bool = True model: str | None = None task_type: str = "text" mcp_servers: list[str] = Field(default_factory=list) class CronPatchBody(BaseModel): + name: str | None = None trigger: str | None = None prompt: str | None = None session_key: str | None = None @@ -43,6 +50,19 @@ def _get_cron_manager(server: Any) -> Any: return server.app_runtime.cron_manager +def _user_may_manage_cron(row: Any, *, agent_row: Any, user: Any) -> bool: + if bool(user.is_admin) or user_owns_agent(agent_row, user): + return True + return bool(row.user_id == user.id) + + +def _assert_cron_access(row: Any, *, agent_id: str, agent_row: Any, user: Any) -> None: + if row.agent_id != agent_id: + raise OctopError(ErrorCode.NOT_FOUND, "cron job not found") + if not _user_may_manage_cron(row, agent_row=agent_row, user=user): + raise OctopError(ErrorCode.FORBIDDEN, "cron job not accessible to user") + + @router.get("/cron/settings", summary="Cron server settings") async def cron_settings( user: Any = Depends(current_user), @@ -59,10 +79,14 @@ async def list_cron( server: Any = Depends(get_server), ) -> list[dict[str, Any]]: """List scheduled jobs for an agent.""" - require_agent_owner_row(agent_id, user=user, as_user=None, server=server) + agent_row = require_agent_row(agent_id, user=user, as_user=None, server=server) + owner_scope = user.is_admin or user_owns_agent(agent_row, user) return [ r.to_public_dict(include_agent=True) - for r in _get_cron_manager(server).list_by_agent(agent_id) + for r in _get_cron_manager(server).list_by_agent( + agent_id, + user_id=None if owner_scope else user.id, + ) ] @@ -77,18 +101,22 @@ async def create_cron( from octop.api.common.validators import validate_chat_mcp_servers # noqa: PLC0415 from octop.infra.cron.manager import CronCreateSpec # noqa: PLC0415 - require_agent_owner_row(agent_id, user=user, as_user=None, server=server) + require_agent_row(agent_id, user=user, as_user=None, server=server) + prompt = require_cron_prompt(body.prompt) + cron_id = new_cron_id() mcp_servers = ( await validate_chat_mcp_servers(server, user_id=user.id, names=body.mcp_servers) or [] ) spec = CronCreateSpec( - cron_id=new_cron_id(), + cron_id=cron_id, agent_id=agent_id, user_id=user.id, + name=require_cron_name(body.name, prompt=prompt, cron_id=cron_id), trigger=body.trigger, - prompt=require_cron_prompt(body.prompt), + prompt=prompt, session_key=body.session_key, fresh_thread=body.fresh_thread, + enabled=body.enabled, model=(body.model or "").strip() or None, task_type=normalize_cron_task_type(body.task_type), mcp_servers=mcp_servers, @@ -106,10 +134,11 @@ async def get_cron( server: Any = Depends(get_server), ) -> dict[str, Any]: """Return one scheduled job by id.""" - require_agent_owner_row(agent_id, user=user, as_user=None, server=server) + agent_row = require_agent_row(agent_id, user=user, as_user=None, server=server) row = _get_cron_manager(server).get(cron_id) - if row is None or row.agent_id != agent_id: + if row is None: raise OctopError(ErrorCode.NOT_FOUND, "cron job not found") + _assert_cron_access(row, agent_id=agent_id, agent_row=agent_row, user=user) return cast(dict[str, Any], row.to_public_dict(include_agent=True)) @@ -125,14 +154,16 @@ async def patch_cron( from octop.api.common.validators import validate_chat_mcp_servers # noqa: PLC0415 from octop.infra.db.repos._base import UNSET # noqa: PLC0415 - require_agent_owner_row(agent_id, user=user, as_user=None, server=server) + agent_row = require_agent_row(agent_id, user=user, as_user=None, server=server) mgr = _get_cron_manager(server) existing = mgr.get(cron_id) - if existing is None or existing.agent_id != agent_id: + if existing is None: raise OctopError(ErrorCode.NOT_FOUND, "cron job not found") + _assert_cron_access(existing, agent_id=agent_id, agent_row=agent_row, user=user) if body.trigger is not None: build_trigger(body.trigger) patch_fields = body.model_dump(exclude_unset=True) + prompt = require_cron_prompt(body.prompt) if body.prompt is not None else None mcp_arg: object = UNSET if "mcp_servers" in patch_fields: mcp_arg = await validate_chat_mcp_servers( @@ -141,7 +172,12 @@ async def patch_cron( row = await mgr.update( cron_id, trigger=body.trigger, - prompt=require_cron_prompt(body.prompt) if body.prompt is not None else None, + name=( + require_cron_name(body.name, prompt=prompt or existing.prompt, cron_id=cron_id) + if "name" in patch_fields + else None + ), + prompt=prompt, session_key=body.session_key, fresh_thread=body.fresh_thread, enabled=int(body.enabled) if body.enabled is not None else None, @@ -160,11 +196,12 @@ async def delete_cron( server: Any = Depends(get_server), ) -> None: """Remove a scheduled job.""" - require_agent_owner_row(agent_id, user=user, as_user=None, server=server) + agent_row = require_agent_row(agent_id, user=user, as_user=None, server=server) mgr = _get_cron_manager(server) existing = mgr.get(cron_id) - if existing is None or existing.agent_id != agent_id: + if existing is None: raise OctopError(ErrorCode.NOT_FOUND, "cron job not found") + _assert_cron_access(existing, agent_id=agent_id, agent_row=agent_row, user=user) await mgr.delete(cron_id) @@ -176,9 +213,10 @@ async def run_now( server: Any = Depends(get_server), ) -> None: """Trigger an immediate one-off run without waiting for the schedule.""" - require_agent_owner_row(agent_id, user=user, as_user=None, server=server) + agent_row = require_agent_row(agent_id, user=user, as_user=None, server=server) mgr = _get_cron_manager(server) existing = mgr.get(cron_id) - if existing is None or existing.agent_id != agent_id: + if existing is None: raise OctopError(ErrorCode.NOT_FOUND, "cron job not found") + _assert_cron_access(existing, agent_id=agent_id, agent_row=agent_row, user=user) await mgr.run_now(cron_id) diff --git a/src/octop/infra/cron/manager.py b/src/octop/infra/cron/manager.py index 59ee4474..132ffad8 100644 --- a/src/octop/infra/cron/manager.py +++ b/src/octop/infra/cron/manager.py @@ -20,7 +20,7 @@ from octop.infra.db.services import RepoBundle from octop.infra.gateway.gateway import Gateway -from octop.infra.cron.task_type import normalize_cron_task_type +from octop.infra.cron.task_type import normalize_cron_task_type, require_cron_name from octop.infra.db.repos._base import UNSET logger = logging.getLogger(__name__) @@ -35,6 +35,7 @@ class CronCreateSpec: user_id: int trigger: str prompt: str + name: str | None = None session_key: str | None = None fresh_thread: bool = False model: str | None = None @@ -115,6 +116,7 @@ async def create(self, spec: CronCreateSpec | None = None, **kwargs: Any) -> Cro cron_id=spec.cron_id, agent_id=spec.agent_id, user_id=spec.user_id, + name=require_cron_name(spec.name, prompt=spec.prompt, cron_id=spec.cron_id), trigger=spec.trigger, prompt=spec.prompt, session_key=session_key, @@ -122,6 +124,7 @@ async def create(self, spec: CronCreateSpec | None = None, **kwargs: Any) -> Cro model=(spec.model or "").strip() or None, task_type=normalize_cron_task_type(spec.task_type), mcp_servers=list(spec.mcp_servers or []), + enabled=spec.enabled, ) row = self._repos.cron_repo.get(spec.cron_id) assert row is not None @@ -139,8 +142,18 @@ async def create(self, spec: CronCreateSpec | None = None, **kwargs: Any) -> Cro def get(self, cron_id: str) -> CronJobRow | None: return self._repos.cron_repo.get(cron_id) - def list_by_agent(self, agent_id: str, *, include_disabled: bool = True) -> list[CronJobRow]: - return self._repos.cron_repo.list_by_agent(agent_id, include_disabled=include_disabled) + def list_by_agent( + self, + agent_id: str, + *, + include_disabled: bool = True, + user_id: int | None = None, + ) -> list[CronJobRow]: + return self._repos.cron_repo.list_by_agent( + agent_id, + include_disabled=include_disabled, + user_id=user_id, + ) def list_all(self, *, include_disabled: bool = True) -> list[CronJobRow]: return self._repos.cron_repo.list_all(include_disabled=include_disabled) @@ -150,6 +163,7 @@ async def update( cron_id: str, *, trigger: str | None = None, + name: str | None = None, prompt: str | None = None, session_key: str | None = None, fresh_thread: bool | None = None, @@ -167,6 +181,7 @@ async def update( raise OctopError(ErrorCode.NOT_FOUND, f"cron job {cron_id!r} not found") repo_kwargs: dict[str, Any] = { "trigger": trigger, + "name": name, "prompt": prompt, "session_key": session_key, "fresh_thread": fresh_thread, @@ -251,6 +266,11 @@ async def _ensure_session(self, session_key: str, *, agent_id: str, user_id: int f"not {agent_id!r}" ) raise ValueError(msg) + if existing.user_id != user_id: + msg = ( + f"session {session_key!r} belongs to user {existing.user_id!r}, not {user_id!r}" + ) + raise ValueError(msg) return parts = session_key.split(":", 3) channel_type = parts[1] if len(parts) >= 2 else "cron" diff --git a/src/octop/infra/cron/task_type.py b/src/octop/infra/cron/task_type.py index 4ee1dfe4..b4081b48 100644 --- a/src/octop/infra/cron/task_type.py +++ b/src/octop/infra/cron/task_type.py @@ -7,6 +7,7 @@ CronTaskType = Literal["text", "agent"] DEFAULT_CRON_TASK_TYPE: CronTaskType = "agent" CRON_PROMPT_MAX_LEN = 2000 +CRON_NAME_MAX_LEN = 80 _CRON_TASK_TYPES = frozenset({"text", "agent"}) @@ -32,3 +33,19 @@ def require_cron_prompt(prompt: str) -> str: if len(text) > CRON_PROMPT_MAX_LEN: raise ValueError(f"prompt must be at most {CRON_PROMPT_MAX_LEN} characters") return text + + +def default_cron_name(prompt: str, cron_id: str) -> str: + """Build a readable fallback name from the prompt or id.""" + text = " ".join(prompt.strip().split()) + if not text: + return cron_id + return text[:40] + + +def require_cron_name(name: str | None, *, prompt: str, cron_id: str) -> str: + """Validate cron display name and fill a stable fallback when omitted.""" + text = (name or "").strip() or default_cron_name(prompt, cron_id) + if len(text) > CRON_NAME_MAX_LEN: + raise ValueError(f"name must be at most {CRON_NAME_MAX_LEN} characters") + return text diff --git a/src/octop/infra/cron/tools.py b/src/octop/infra/cron/tools.py index e4a2b740..429880a1 100644 --- a/src/octop/infra/cron/tools.py +++ b/src/octop/infra/cron/tools.py @@ -11,6 +11,7 @@ from octop.infra.cron.manager import CronCreateSpec from octop.infra.cron.task_type import ( + require_cron_name, require_cron_prompt, require_cron_task_type, ) @@ -117,6 +118,10 @@ async def cronjob_create( ), ), ], + name: Annotated[ + str | None, + Field(description="Optional display name for this cron job."), + ] = None, fresh_thread: Annotated[ bool, Field(description="If true, reset conversation context before each agent run."), @@ -132,12 +137,15 @@ async def cronjob_create( ) -> str: try: agent_id, user_id, session_key = _tool_ctx() + cron_id = new_cron_id() + cleaned_prompt = require_cron_prompt(prompt) spec = CronCreateSpec( - cron_id=new_cron_id(), + cron_id=cron_id, agent_id=agent_id, user_id=user_id, + name=require_cron_name(name, prompt=cleaned_prompt, cron_id=cron_id), trigger=trigger, - prompt=require_cron_prompt(prompt), + prompt=cleaned_prompt, fresh_thread=fresh_thread, session_key=session_key, enabled=enabled, @@ -150,6 +158,7 @@ async def cronjob_create( async def cronjob_update( cron_id: str, + name: Annotated[str | None, Field(description="New display name.")] = None, trigger: Annotated[str | None, Field(description=f"New schedule. {_TRIGGER_HELP}")] = None, prompt: Annotated[ str | None, @@ -161,11 +170,21 @@ async def cronjob_update( ) -> str: try: agent_id, user_id, _session_key = _tool_ctx() - _get_owned(mgr, cron_id, agent_id, user_id) + existing = _get_owned(mgr, cron_id, agent_id, user_id) + cleaned_prompt = require_cron_prompt(prompt) if prompt is not None else None row = await mgr.update( cron_id, + name=( + require_cron_name( + name, + prompt=cleaned_prompt or existing.prompt, + cron_id=cron_id, + ) + if name is not None + else None + ), trigger=trigger, - prompt=require_cron_prompt(prompt) if prompt is not None else None, + prompt=cleaned_prompt, fresh_thread=fresh_thread, enabled=int(enabled) if enabled is not None else None, task_type=require_cron_task_type(task_type) if task_type is not None else None, diff --git a/src/octop/infra/db/migrate.py b/src/octop/infra/db/migrate.py index 0343dc42..48d1c75f 100644 --- a/src/octop/infra/db/migrate.py +++ b/src/octop/infra/db/migrate.py @@ -143,6 +143,24 @@ def _ensure_agent_profile_columns(db: DatabasePool) -> None: _collapse_legacy_agent_welcome_columns(db) +def _ensure_cron_jobs_schema(db: DatabasePool) -> None: + if not _table_exists(db, "cron_jobs"): + return + _ensure_column(db, "cron_jobs", "name", "TEXT NOT NULL DEFAULT ''") + _ensure_column( + db, + "cron_jobs", + "task_type", + "TEXT NOT NULL DEFAULT 'agent' CHECK (task_type IN ('text', 'agent'))", + ) + _ensure_column( + db, + "cron_jobs", + "mcp_servers", + "TEXT NOT NULL DEFAULT '[]'", + ) + + def _collapse_legacy_agent_welcome_columns(db: DatabasePool) -> None: """Fold welcome_message_zh/en into a single instance-owned welcome_message.""" if not _table_exists(db, "agents"): @@ -926,19 +944,7 @@ def _repair_legacy_schema(db: DatabasePool) -> None: # Pre-squash DBs may already report version ≥6; reconcile can clamp to 6 # without applying 006_user_permissions.sql — ensure the column here. _ensure_column(db, "users", "permissions", "TEXT NOT NULL DEFAULT '[]'") - if _table_exists(db, "cron_jobs"): - _ensure_column( - db, - "cron_jobs", - "task_type", - "TEXT NOT NULL DEFAULT 'agent' CHECK (task_type IN ('text', 'agent'))", - ) - _ensure_column( - db, - "cron_jobs", - "mcp_servers", - "TEXT NOT NULL DEFAULT '[]'", - ) + _ensure_cron_jobs_schema(db) if _table_exists(db, "threads"): _ensure_column(db, "threads", "model_ref", "TEXT") _ensure_column(db, "threads", "reasoning_mode", "TEXT") @@ -996,6 +1002,8 @@ def _reconcile_pre_squash_schema_version(db: DatabasePool) -> None: _ensure_user_invites_schema(db) if max_version >= 10: _ensure_thread_message_projection_schema(db) + if max_version >= 11: + _ensure_cron_jobs_schema(db) with db.connect() as conn: conn.execute("UPDATE _schema_version SET version = %s", (max_version,)) return @@ -1025,6 +1033,8 @@ def _reconcile_pre_squash_schema_version(db: DatabasePool) -> None: _ensure_user_invites_schema(db) if max_version >= 10: _ensure_thread_message_projection_schema(db) + if max_version >= 11: + _ensure_cron_jobs_schema(db) with db.connect() as conn: conn.execute("UPDATE _schema_version SET version = ?", (max_version,)) @@ -1052,10 +1062,11 @@ def _apply_sqlite_migration(db: DatabasePool, version: int, path: Path) -> None: (idempotent); PostgreSQL runs the ``.pg.sql`` file then the same helpers as a no-op safety net. ``config_json`` profile keys are backfilled in Python either way. - Version 8 adds cache-aware usage buckets and model call counts. - Version 9 adds one-time ``user_invites`` codes. - Version 10 adds the dashboard thread-message projection when the legacy - database actually contains conversation tables. + Version 8 adds cache-aware usage buckets and model call counts. + Version 9 adds one-time ``user_invites`` codes. + Version 10 adds the dashboard thread-message projection when the legacy + database actually contains conversation tables. + Version 11 adds cron job display names. """ if version == 2: if _table_exists(db, "cron_jobs"): @@ -1134,6 +1145,11 @@ def _apply_sqlite_migration(db: DatabasePool, version: int, path: Path) -> None: with db.connect() as conn: conn.execute("UPDATE _schema_version SET version = ?", (version,)) return + if version == 11: + _ensure_cron_jobs_schema(db) + with db.connect() as conn: + conn.execute("UPDATE _schema_version SET version = ?", (version,)) + return sql = path.read_text(encoding="utf-8") with db.connect() as conn: conn.executescript(sql) @@ -1168,3 +1184,4 @@ def run_migrations(db: DatabasePool) -> None: _ensure_usage_cache_schema(db) _ensure_user_invites_schema(db) _ensure_thread_message_projection_schema(db) + _ensure_cron_jobs_schema(db) diff --git a/src/octop/infra/db/migrations/011_cron_job_names.pg.sql b/src/octop/infra/db/migrations/011_cron_job_names.pg.sql new file mode 100644 index 00000000..cddbb60f --- /dev/null +++ b/src/octop/infra/db/migrations/011_cron_job_names.pg.sql @@ -0,0 +1,5 @@ +-- Schema v11: cron job display names. + +ALTER TABLE cron_jobs ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''; + +UPDATE _schema_version SET version = 11; diff --git a/src/octop/infra/db/migrations/011_cron_job_names.sql b/src/octop/infra/db/migrations/011_cron_job_names.sql new file mode 100644 index 00000000..cd1cbbe4 --- /dev/null +++ b/src/octop/infra/db/migrations/011_cron_job_names.sql @@ -0,0 +1,7 @@ +-- Schema v11: cron job display names. +-- SQLite applies this via migrate.py::_ensure_cron_jobs_schema so the upgrade +-- remains idempotent for databases repaired ahead of the version watermark. + +ALTER TABLE cron_jobs ADD COLUMN name TEXT NOT NULL DEFAULT ''; + +UPDATE _schema_version SET version = 11; diff --git a/src/octop/infra/db/repos/cron.py b/src/octop/infra/db/repos/cron.py index 3e9aa43d..0af5cb65 100644 --- a/src/octop/infra/db/repos/cron.py +++ b/src/octop/infra/db/repos/cron.py @@ -9,6 +9,7 @@ from octop.infra.cron.task_type import ( DEFAULT_CRON_TASK_TYPE, CronTaskType, + default_cron_name, normalize_cron_task_type, require_cron_task_type, ) @@ -51,10 +52,20 @@ def _decode_mcp_servers(raw: Any) -> list[str]: return [str(n).strip() for n in parsed if str(n).strip()] +def _optional_str(r: DbRow, key: str) -> str | None: + try: + value = r[key] + except (KeyError, IndexError): + return None + text = str(value).strip() if value else "" + return text or None + + @dataclass(frozen=True) class CronJobRow: id: int cron_id: str + name: str agent_id: str user_id: int trigger: str @@ -76,6 +87,7 @@ def from_row(cls, r: DbRow) -> CronJobRow: return cls( id=r["id"], cron_id=r["cron_id"], + name=_optional_str(r, "name") or default_cron_name(r["prompt"], r["cron_id"]), agent_id=r["agent_id"], user_id=r["user_id"], trigger=r["schedule_spec"], @@ -96,6 +108,7 @@ def from_row(cls, r: DbRow) -> CronJobRow: def to_public_dict(self, *, include_agent: bool = False) -> dict[str, Any]: out: dict[str, Any] = { "id": self.cron_id, + "name": self.name, "trigger": self.trigger, "prompt": self.prompt, "session_key": self.session_key, @@ -130,16 +143,19 @@ def create( model: str | None = None, task_type: CronTaskType = DEFAULT_CRON_TASK_TYPE, mcp_servers: list[str] | None = None, + enabled: bool = True, + name: str | None = None, ) -> str: ts = now_ts() with self._db.transaction() as conn: conn.execute( - "INSERT INTO cron_jobs(cron_id, agent_id, user_id, schedule_spec, prompt, " + "INSERT INTO cron_jobs(cron_id, name, agent_id, user_id, schedule_spec, prompt, " "session_key, model, fresh_thread, task_type, mcp_servers, enabled, " "created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)", + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( cron_id, + (name or "").strip() or default_cron_name(prompt, cron_id), agent_id, user_id, trigger, @@ -149,6 +165,7 @@ def create( bool_int(fresh_thread), task_type, _encode_mcp_servers(mcp_servers), + bool_int(enabled), ts, ts, ), @@ -160,13 +177,23 @@ def get(self, cron_id: str) -> CronJobRow | None: r = conn.execute("SELECT * FROM cron_jobs WHERE cron_id = ?", (cron_id,)).fetchone() return CronJobRow.from_row(r) if r else None - def list_by_agent(self, agent_id: str, *, include_disabled: bool = True) -> list[CronJobRow]: + def list_by_agent( + self, + agent_id: str, + *, + include_disabled: bool = True, + user_id: int | None = None, + ) -> list[CronJobRow]: sql = "SELECT * FROM cron_jobs WHERE agent_id = ?" + params: list[object] = [agent_id] + if user_id is not None: + sql += " AND user_id = ?" + params.append(user_id) if not include_disabled: sql += " AND enabled = 1" sql += " ORDER BY created_at DESC" with self._db.connect() as conn: - rows = conn.execute(sql, (agent_id,)).fetchall() + rows = conn.execute(sql, params).fetchall() return map_rows(rows, CronJobRow) def list_all(self, *, include_disabled: bool = True) -> list[CronJobRow]: @@ -198,6 +225,7 @@ def update( cron_id: str, *, trigger: str | None = None, + name: str | None = None, prompt: str | None = None, session_key: str | None = None, fresh_thread: bool | None = None, @@ -209,6 +237,7 @@ def update( fields, params = partial_updates( [ ("schedule_spec", trigger), + ("name", name), ("prompt", prompt), ("session_key", session_key), ("task_type", task_type), diff --git a/tests/integration/test_cron_shared_agent.py b/tests/integration/test_cron_shared_agent.py new file mode 100644 index 00000000..635d317d --- /dev/null +++ b/tests/integration/test_cron_shared_agent.py @@ -0,0 +1,51 @@ +"""Cron jobs on shared agents are scoped to the acting user.""" + +from __future__ import annotations + +from tests.support.auth import create_agent, create_user, seed_openai_provider + + +async def test_peer_can_create_own_cron_on_shared_agent(env) -> None: + client, _srv, admin_auth = env + await seed_openai_provider(client, admin_auth) + owner_auth = await create_user(client, admin_auth, username="cron_owner") + peer_auth = await create_user(client, admin_auth, username="cron_peer") + other_auth = await create_user(client, admin_auth, username="cron_other") + agent_id = await create_agent(client, owner_auth, name="shared-cron-bot") + + response = await client.patch( + f"/api/agents/{agent_id}", + headers=owner_auth, + json={"is_shared": True}, + ) + assert response.status_code == 200, response.text + + response = await client.post( + f"/api/agents/{agent_id}/cron", + headers=peer_auth, + json={ + "name": "Peer morning brief", + "trigger": "cron:0 9 * * *", + "prompt": "Summarize today's calendar", + "enabled": False, + "task_type": "agent", + }, + ) + assert response.status_code == 201, response.text + created = response.json() + assert created["name"] == "Peer morning brief" + assert created["enabled"] is False + + response = await client.get(f"/api/agents/{agent_id}/cron", headers=peer_auth) + assert response.status_code == 200, response.text + assert [row["id"] for row in response.json()] == [created["id"]] + + response = await client.get(f"/api/agents/{agent_id}/cron", headers=other_auth) + assert response.status_code == 200, response.text + assert response.json() == [] + + response = await client.get( + f"/api/agents/{agent_id}/cron/{created['id']}", + headers=other_auth, + ) + assert response.status_code == 403 diff --git a/tests/unit/cron/test_cron_manager.py b/tests/unit/cron/test_cron_manager.py index 77d515ca..5b600652 100644 --- a/tests/unit/cron/test_cron_manager.py +++ b/tests/unit/cron/test_cron_manager.py @@ -233,11 +233,13 @@ async def test_create_persists_row(tmp_path: Path) -> None: cron_id=cid, agent_id=aid, user_id=uid, + name="Morning report", trigger="cron:0 9 * * *", prompt="morning report", ) assert row.cron_id == cid + assert row.name == "Morning report" assert row.prompt == "morning report" assert row.agent_id == aid assert mgr.get(cid) is not None @@ -261,6 +263,25 @@ async def test_create_schedules_job(tmp_path: Path) -> None: mgr._scheduler.add_job.assert_called_once() +@pytest.mark.asyncio +async def test_create_can_start_disabled(tmp_path: Path) -> None: + services = _make_services(tmp_path) + aid, uid = _make_agent(services) + mgr = _make_manager(services) + + row = await mgr.create( + cron_id=_cron_id(), + agent_id=aid, + user_id=uid, + trigger="interval:30", + prompt="ping", + enabled=False, + ) + + assert row.enabled == 0 + mgr._scheduler.add_job.assert_not_called() + + @pytest.mark.asyncio async def test_create_writes_audit_entry(tmp_path: Path) -> None: """create() writes a cron.create audit log entry.""" diff --git a/tests/unit/db/test_agent_profile_columns.py b/tests/unit/db/test_agent_profile_columns.py index 1b373b61..a555eb8b 100644 --- a/tests/unit/db/test_agent_profile_columns.py +++ b/tests/unit/db/test_agent_profile_columns.py @@ -110,7 +110,7 @@ def test_migration_007_backfills_profile_columns(tmp_path: Path) -> None: with pool.connect() as conn: version = conn.execute("SELECT version FROM _schema_version").fetchone()[0] row = conn.execute("SELECT * FROM agents WHERE agent_id = ?", ("ag1",)).fetchone() - assert version == 10 + assert version == 11 assert row["template_name"] == "general-assistant" assert row["icon_name"] == "zap" assert row["icon_url"] == "https://cdn.example.com/a.png" diff --git a/tests/unit/db/test_clip_thread_title.py b/tests/unit/db/test_clip_thread_title.py index d916cb1b..893ee5b4 100644 --- a/tests/unit/db/test_clip_thread_title.py +++ b/tests/unit/db/test_clip_thread_title.py @@ -82,7 +82,7 @@ def test_migration_003_repairs_stored_hard_cuts(tmp_path: Path) -> None: with pool.connect() as conn: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] title = conn.execute("SELECT title FROM threads WHERE thread_id = ?", ("t1",)).fetchone()[0] - assert v == 10 + assert v == 11 assert title == "x" * 39 + "…" # Idempotent repair assert repair_all_legacy_thread_titles(pool) == 0 diff --git a/tests/unit/db/test_db_pool.py b/tests/unit/db/test_db_pool.py index 1fa6f505..ffe85d66 100644 --- a/tests/unit/db/test_db_pool.py +++ b/tests/unit/db/test_db_pool.py @@ -85,7 +85,7 @@ def test_run_migrations_idempotent(db: SqlitePool): doc_cols = { r["name"] for r in conn.execute("PRAGMA table_info(knowledge_documents)").fetchall() } - assert v == 10 + assert v == 11 assert "login_failed_count" in cols assert "login_locked_until" in cols assert "preferences_json" in cols @@ -95,6 +95,7 @@ def test_run_migrations_idempotent(db: SqlitePool): assert {"thread_messages", "thread_history_projection"}.issubset(table_names) assert "task_type" in cron_cols assert "mcp_servers" in cron_cols + assert "name" in cron_cols assert {"model_ref", "reasoning_mode", "reasoning_effort", "artifacts"}.issubset(thread_cols) assert { "color", @@ -126,6 +127,7 @@ def test_repair_legacy_schema_ensures_columns(tmp_path: Path) -> None: with pool.connect() as conn: cron_cols = {r["name"] for r in conn.execute("PRAGMA table_info(cron_jobs)").fetchall()} assert "task_type" in cron_cols + assert "name" in cron_cols def test_migration_002_idempotent_when_column_already_present(tmp_path: Path) -> None: @@ -144,7 +146,7 @@ def test_migration_002_idempotent_when_column_already_present(tmp_path: Path) -> with pool.connect() as conn: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] cron_cols = {r["name"] for r in conn.execute("PRAGMA table_info(cron_jobs)").fetchall()} - assert v == 10 + assert v == 11 assert "mcp_servers" in cron_cols assert "skill_packages" in { r["name"] @@ -281,7 +283,7 @@ def test_stuck_version_6_without_permissions_column_is_repaired(tmp_path: Path) with pool.connect() as conn: cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()} version = conn.execute("SELECT version FROM _schema_version").fetchone()[0] - assert version == 10 + assert version == 11 assert "permissions" in cols @@ -305,9 +307,11 @@ def test_schema_v10_without_projection_tables_is_repaired(tmp_path: Path) -> Non for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() } kb_cols = {r["name"] for r in conn.execute("PRAGMA table_info(knowledge_bases)").fetchall()} - assert version == 10 + cron_cols = {r["name"] for r in conn.execute("PRAGMA table_info(cron_jobs)").fetchall()} + assert version == 11 assert {"thread_messages", "thread_history_projection"}.issubset(table_names) assert "max_documents" in kb_cols + assert "name" in cron_cols def test_ahead_of_max_schema_version_clamps_to_max(tmp_path: Path) -> None: @@ -337,7 +341,7 @@ def test_ahead_of_max_schema_version_clamps_to_max(tmp_path: Path) -> None: r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() } - assert version == 10 + assert version == 11 assert "skill_package_id" in pkg_cols assert "published_expert_id" in pub_cols assert "user_invites" in invite_tables @@ -376,7 +380,7 @@ def test_pre_squash_schema_version_clamped_and_knowledge_tables_filled( for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() } user_cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()} - assert version == 10 + assert version == 11 assert "permissions" in user_cols assert { "published_experts", diff --git a/tests/unit/db/test_published_experts_repo.py b/tests/unit/db/test_published_experts_repo.py index 4705d7f2..226e5a0a 100644 --- a/tests/unit/db/test_published_experts_repo.py +++ b/tests/unit/db/test_published_experts_repo.py @@ -27,7 +27,7 @@ def test_published_experts_table_exists(db: SqlitePool) -> None: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] cols = {r["name"] for r in conn.execute("PRAGMA table_info(published_experts)").fetchall()} assert "published_experts" in names - assert v == 10 + assert v == 11 assert "published_expert_id" in cols diff --git a/tests/unit/db/test_repo_cron.py b/tests/unit/db/test_repo_cron.py index 94ba285a..dc964171 100644 --- a/tests/unit/db/test_repo_cron.py +++ b/tests/unit/db/test_repo_cron.py @@ -51,6 +51,7 @@ def test_create_and_get(repo: CronJobRepo, agent_id: str, user_id: int): cron_id=cid, agent_id=agent_id, user_id=user_id, + name="Morning report", trigger="0 9 * * *", prompt="run report", session_key=_session_key(agent_id, user_id), @@ -58,10 +59,12 @@ def test_create_and_get(repo: CronJobRepo, agent_id: str, user_id: int): row = repo.get(cid) assert isinstance(row, CronJobRow) assert row.cron_id == cid + assert row.name == "Morning report" assert row.trigger == "0 9 * * *" assert row.enabled == 1 assert row.task_type == "agent" assert row.last_run_at is None + assert row.to_public_dict()["name"] == "Morning report" def test_create_with_task_type(repo: CronJobRepo, agent_id: str, user_id: int): @@ -129,6 +132,36 @@ def test_list_by_agent(repo: CronJobRepo, agent_id: str, user_id: int): assert len(rows) == 2 +def test_list_by_agent_can_filter_user( + db: SqlitePool, + repo: CronJobRepo, + agent_id: str, + user_id: int, +): + other_user = UserRepo(db).create(username="bob", password_hash="h", role="user") + repo.create( + cron_id=new_ulid(), + agent_id=agent_id, + user_id=user_id, + trigger="* * * * *", + prompt="mine", + session_key=_session_key(agent_id, user_id), + ) + repo.create( + cron_id=new_ulid(), + agent_id=agent_id, + user_id=other_user, + trigger="* * * * *", + prompt="theirs", + session_key=_session_key(agent_id, other_user), + ) + + rows = repo.list_by_agent(agent_id, user_id=user_id) + + assert len(rows) == 1 + assert rows[0].prompt == "mine" + + def test_update_partial(repo: CronJobRepo, agent_id: str, user_id: int): cid = new_ulid() repo.create( diff --git a/tests/unit/db/test_repo_knowledge.py b/tests/unit/db/test_repo_knowledge.py index 00d38a4f..b0ec9312 100644 --- a/tests/unit/db/test_repo_knowledge.py +++ b/tests/unit/db/test_repo_knowledge.py @@ -50,7 +50,7 @@ def test_knowledge_tables_migrated(db: SqlitePool) -> None: "knowledge_bases", "knowledge_documents", }.issubset(names) - assert v == 10 + assert v == 11 assert "knowledge_base_members" not in names cols = {r["name"] for r in conn.execute("PRAGMA table_info(knowledge_bases)").fetchall()} assert "knowledge_base_id" in cols diff --git a/tests/unit/db/test_skill_package_icons.py b/tests/unit/db/test_skill_package_icons.py index 27ac3646..4a6e3999 100644 --- a/tests/unit/db/test_skill_package_icons.py +++ b/tests/unit/db/test_skill_package_icons.py @@ -91,7 +91,7 @@ def test_migration_002_idempotent_when_icon_columns_already_present(tmp_path: Pa "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='skill_packages'" ) } - assert v == 10 + assert v == 11 assert "icon_name" in cols assert "icon_url" in cols assert "skill_package_id" in cols @@ -110,7 +110,7 @@ def test_repair_legacy_schema_adds_icon_columns_at_version_2(tmp_path: Path) -> with pool.connect() as conn: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] cols = {r["name"] for r in conn.execute("PRAGMA table_info(skill_packages)").fetchall()} - assert v == 10 + assert v == 11 assert "icon_name" in cols assert "icon_url" in cols assert "skill_package_id" in cols diff --git a/tests/unit/db/test_skill_packages_repo.py b/tests/unit/db/test_skill_packages_repo.py index 67d7cc58..e56acde8 100644 --- a/tests/unit/db/test_skill_packages_repo.py +++ b/tests/unit/db/test_skill_packages_repo.py @@ -28,7 +28,7 @@ def test_skill_packages_table_exists(db: SqlitePool) -> None: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] cols = {r["name"] for r in conn.execute("PRAGMA table_info(skill_packages)").fetchall()} assert "skill_packages" in names - assert v == 10 + assert v == 11 assert "skill_package_id" in cols