feat: add schedule kanban

Signed-off-by: Chojan Shang <psiace@apache.org>

Author: PsiACE

contrib/bubseek-schedule/README.md

@@ -33,12 +33,24 @@ dependencies = [
 
 ## Runtime Behavior
 
-- Scheduler starts when the plugin channel starts.
+- The plugin uses **APScheduler `BackgroundScheduler`** (see also upstream [bub-schedule](https://github.com/bubbuild/bub-contrib) JSON store pattern: persistence must not depend on a specific channel being enabled).
+- **`load_state` starts the scheduler** on the first inbound message. That way `bub chat` (CLI-only: only the `cli` channel is enabled) still persists jobs to SeekDB. Previously, `AsyncIOScheduler` was only started by the `schedule` channel, so CLI chat left jobs in memory-only `_pending_jobs` and **nothing was written to `apscheduler_jobs`**.
+- The channel name is `schedule`. Enabling it in `bub gateway` is optional for persistence; it still starts/stops the scheduler cleanly when you use gateway with that channel.
 - Jobs are persisted to:
-  - **OceanBase/SeekDB**: When `BUB_TAPESTORE_SQLALCHEMY_URL` uses `mysql+oceanbase`, jobs go to the same database (table `apscheduler_jobs`).
+  - **OceanBase/SeekDB**: Same URL as the tape store (`BUB_TAPESTORE_SQLALCHEMY_URL` / `OCEANBASE_*`), table `apscheduler_jobs`.
 
 ## Provided Tools
 
 - `schedule.add`: Add a scheduled job with cron, interval, or one-shot.
 - `schedule.remove`: Remove a scheduled job by ID.
 - `schedule.list`: List all scheduled jobs.
+
+## Debug: job in chat but not in Marimo kanban / DB
+
+The gateway resolves the job store URL from `BUB_TAPESTORE_SQLALCHEMY_URL` or workspace `.env` (`OCEANBASE_*`). Marimo must use the **same** URL. If `insights/schedule_kanban.py` pointed at the default `127.0.0.1:2881/bub` while your `.env` uses another host/db, the table will look empty.
+
+From the bubseek repo root:
+
+```bash
+uv run python scripts/query_apscheduler_jobs.py --job-id <id>
+```

contrib/bubseek-schedule/src/bubseek_schedule/channel.py

@@ -1,21 +1,33 @@
 import asyncio
+import contextlib
 from asyncio import Event
+from typing import ClassVar
 
+from apscheduler.schedulers import SchedulerAlreadyRunningError
 from apscheduler.schedulers.base import BaseScheduler
 from bub.channels import Channel
 from loguru import logger
 
 
 class ScheduleChannel(Channel):
+    """Starts/stops BackgroundScheduler when this channel is enabled (e.g. gateway)."""
+
+    name: ClassVar[str] = "schedule"
+
     def __init__(self, scheduler: BaseScheduler) -> None:
         self.scheduler = scheduler
 
     async def start(self, stop_event: Event) -> None:
-        loop = asyncio.get_running_loop()
-        loop.call_soon_threadsafe(self.scheduler.start)
+        if not self.scheduler.running:
+            with contextlib.suppress(SchedulerAlreadyRunningError):
+                self.scheduler.start()
         logger.info("schedule.start complete")
 
     async def stop(self) -> None:
+        if not self.scheduler.running:
+            logger.info("schedule.stop complete (idle)")
+            return
+        # BackgroundScheduler.shutdown() blocks until the worker thread stops.
         loop = asyncio.get_running_loop()
-        loop.call_soon_threadsafe(self.scheduler.shutdown)
+        await loop.run_in_executor(None, lambda: self.scheduler.shutdown(wait=True))
         logger.info("schedule.stop complete")

contrib/bubseek-schedule/src/bubseek_schedule/jobs.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from pathlib import Path
 
 from bub import BubFramework
@@ -13,7 +14,7 @@ def _noop() -> None:
     pass
 
 
-async def run_scheduled_reminder(message: str, session_id: str, workspace: str | None = None) -> None:
+async def _run_scheduled_reminder_async(message: str, session_id: str, workspace: str | None = None) -> None:
     framework = BubFramework()
     framework.load_hooks()
     if workspace:
@@ -30,3 +31,12 @@ async def run_scheduled_reminder(message: str, session_id: str, workspace: str |
         chat_id=chat_id,
     )
     await framework.process_inbound(payload)
+
+
+def run_scheduled_reminder(message: str, session_id: str, workspace: str | None = None) -> None:
+    """Synchronous job target for BackgroundScheduler (runs in a worker thread).
+
+    APScheduler's BackgroundScheduler executes jobs in a thread pool; ``async def`` targets are
+    not supported with the default executor. Each run uses ``asyncio.run()`` with a fresh loop.
+    """
+    asyncio.run(_run_scheduled_reminder_async(message, session_id, workspace))

contrib/bubseek-schedule/src/bubseek_schedule/jobstore.py

@@ -15,12 +15,13 @@
 from sqlalchemy.orm.session import sessionmaker as sessionmaker_type
 
 import bubseek.oceanbase  # noqa: F401 - register mysql+oceanbase dialect
-from bubseek.config import BubSeekSettings
 
 
 def _get_jobstore_url() -> str:
-    """Get SQLAlchemy URL for job store from the configured SeekDB/OceanBase tapestore."""
-    return BubSeekSettings().db.resolved_tapestore_url
+    """Resolve tapestore URL (workspace .env, BUB_WORKSPACE_PATH, cwd) like the rest of bubseek."""
+    from bubseek.config import resolve_tapestore_url
+
+    return resolve_tapestore_url()
 
 
 def _normalize_url(url: str) -> str:
@@ -41,17 +42,22 @@ class OceanBaseJobStore(BaseJobStore):
 
     def __init__(self, url: str | None = None, tablename: str = "apscheduler_jobs"):
         super().__init__()
-        self._url = _normalize_url(url or _get_jobstore_url())
+        self._url_explicit = url
         self._tablename = tablename
         self._engine: Engine | None = None
         self._session_factory: sessionmaker_type | None = None
         self._table: Table | None = None
         self._lock = threading.RLock()
 
+    def _connection_url(self) -> str:
+        if self._url_explicit is not None:
+            return _normalize_url(self._url_explicit)
+        return _normalize_url(_get_jobstore_url())
+
     def _ensure_initialized(self) -> None:
         if self._engine is not None and self._session_factory is not None and self._table is not None:
             return
-        self._engine = create_engine(self._url, pool_pre_ping=True)
+        self._engine = create_engine(self._connection_url(), pool_pre_ping=True)
         self._session_factory = sessionmaker(bind=self._engine, expire_on_commit=False)
         self._init_table()
 

contrib/bubseek-schedule/src/bubseek_schedule/plugin.py

@@ -1,4 +1,7 @@
-from apscheduler.schedulers.asyncio import AsyncIOScheduler
+import contextlib
+
+from apscheduler.schedulers import SchedulerAlreadyRunningError
+from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.schedulers.base import BaseScheduler
 from bub import hookimpl
 from bub.types import Envelope, MessageHandler, State
@@ -8,17 +11,32 @@
 
 def default_scheduler() -> BaseScheduler:
     job_store = OceanBaseJobStore()
-    return AsyncIOScheduler(jobstores={"default": job_store})
+    return BackgroundScheduler(jobstores={"default": job_store})
 
 
 class ScheduleImpl:
+    """Schedule plugin: persist jobs to SeekDB via OceanBaseJobStore.
+
+    Uses BackgroundScheduler so the scheduler can start without the ``schedule`` channel.
+    ``bub chat`` only enables the ``cli`` channel; previously AsyncIOScheduler never started,
+    so APScheduler kept jobs in memory-only ``_pending_jobs`` and nothing reached the DB.
+    """
+
     def __init__(self) -> None:
         from bubseek_schedule import tools  # noqa: F401
 
         self.scheduler = default_scheduler()
 
+    def _ensure_scheduler_started(self) -> None:
+        if self.scheduler.running:
+            return
+        with contextlib.suppress(SchedulerAlreadyRunningError):
+            self.scheduler.start()
+
     @hookimpl
     def load_state(self, message: Envelope, session_id: str) -> State:
+        # Runs before tools on every inbound message — covers CLI-only ``bub chat``.
+        self._ensure_scheduler_started()
         return {"scheduler": self.scheduler}
 
     @hookimpl