fix: activate flow schedules in the Scheduler Service (not just write the config)

.claude-plugin/marketplace.json

@@ -10,7 +10,7 @@
   "plugins": [
     {
       "name": "kbagent",
-      "version": "0.66.0",
+      "version": "0.66.1",
       "source": "./plugins/kbagent",
       "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces",
       "category": "development"

plugins/kbagent/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "kbagent",
-  "version": "0.66.0",
+  "version": "0.66.1",
   "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces",
   "author": {
     "name": "Keboola",

plugins/kbagent/skills/kbagent/references/commands-reference.md

@@ -225,8 +225,8 @@ Requires the project to be added with its **master ('owner') Storage API token**
 - `flow new --project NAME --name NAME [--description D] [--file @path.yaml|-|JSON] [--branch ID]` -- create a conditional flow; validated against the **live** CF schema fetched from the stack before the API call (`INVALID_FLOW_DEFINITION` on failure). A schema-fetch failure does NOT block the write: structural check skipped, semantic checks still run, a `structural schema validation skipped` warning is surfaced
 - `flow update --project NAME --flow-id ID [--name N] [--description D] [--file @path.yaml|-|JSON] [--branch ID]` -- update name, description, or phases/tasks; `--file` is a full-replace of phases+tasks; merge-aware validation against the live CF schema (same graceful semantic-only degradation on fetch failure); requires at least one of --name/--description/--file
 - `flow delete --project NAME --flow-id ID [--branch ID] [--yes]` -- delete a flow config (confirmation guard)
-- `flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--timezone TZ] [--disabled] [--branch ID]` -- attach a cron schedule (stored as keboola.scheduler config, target.componentId=keboola.flow); replaces any existing schedule
-- `flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes]` -- remove all cron schedules attached to a flow; idempotent
+- `flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--timezone TZ] [--disabled] [--branch ID]` -- attach a cron schedule (stored as keboola.scheduler config, target.componentId=keboola.flow); replaces any existing schedule; on production it also ACTIVATES the schedule in the Scheduler Service so it actually fires (without activation the config never runs) -- see gotchas.md
+- `flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes]` -- remove all cron schedules attached to a flow; deactivates the Scheduler Service registration then deletes the config; idempotent
 
 ## Schedule Discovery & Audit (Fleet-Wide)
 - `schedule list [--project NAME ...] [--enabled-only] [--branch ID]` -- fleet-wide list of every `keboola.scheduler` config across one, many, or all projects (parallel fan-out, no --project = all). Each row has `project_alias`, `schedule_id`, `schedule_name`, `parent_component_id`, `parent_config_id`, `parent_name`, `cron`, `timezone`, `enabled`. Answers "which configs are running on cron triggers across N projects?" without enumerating flows

plugins/kbagent/skills/kbagent/references/gotchas.md

@@ -2284,6 +2284,26 @@ The trade-off is deliberate: one big call avoids the O(unique-parents) round-tri
   scheduler configs that target the flow. Pair it with `--dry-run` to see the
   affected configs (cron + timezone) without calling `delete_config`.
 
+## Flow: `schedule` ACTIVATES in the Scheduler Service, not just writes the config (since v0.66.1)
+
+- A `keboola.scheduler` config with `state=enabled` does **NOT** fire on its
+  own. The cron engine lives in a separate Scheduler microservice
+  (`scheduler.{stack}`), which only runs schedules that have been *activated*
+  (`POST /schedules {configurationId}`) — activation registers the schedule and
+  mints its scheduler token. Writing the Storage config is only half the job.
+- Before v0.66.1 `flow schedule` created the config but never activated it, so
+  the schedule looked "enabled" in `schedule list`/`detail` (which just read the
+  config field) yet never ran. v0.66.1 activates after the config write; the
+  JSON result gains `activated` + `activated_schedule_id`.
+- Activation is production-only. On `--branch`, the config is written but NOT
+  activated (`activated=false`) — dev-branch schedules activate when the branch
+  is deployed to production. This mirrors how the Scheduler Service operates on
+  the default branch only.
+- `flow schedule-remove` symmetrically **deactivates** (`DELETE /schedules/{id}`)
+  each matching activation before deleting the config, so removal never orphans
+  an activation + token that would keep firing against a deleted config. The
+  JSON result gains `deactivated_schedule_ids`.
+
 ## `search` is a top-level command, not `config search` (since v0.30.0)
 
 `kbagent search QUERY` searches across **all item types** (tables, buckets, configs, flows, data apps, transformations) via the Storage API global-search endpoint. It is distinct from `kbagent config search --query Q` which scans only configuration JSON bodies.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "keboola-cli"
-version = "0.66.0"
+version = "0.66.1"
 description = "AI-friendly CLI for managing Keboola projects"
 readme = "README.md"
 requires-python = ">=3.12"

src/keboola_agent_cli/changelog.py

@@ -24,6 +24,15 @@
 
 # Ordered newest-first.  Each value is a list of brief one-line descriptions.
 CHANGELOG: dict[str, list[str]] = {
+    "0.66.1": [
+        "Fix: `flow schedule` now ACTIVATES the schedule in the Scheduler Service, not just "
+        "writes the keboola.scheduler config. Previously the command created the config with "
+        "state=enabled but never called the Scheduler microservice, so the schedule was never "
+        "registered and never fired -- matching the UI, which does both steps. Activation runs "
+        "only on production (dev-branch schedules activate on deploy). `flow schedule-remove` "
+        "symmetrically deactivates the Scheduler Service registration before deleting the "
+        "config, so removal no longer orphans an activation (and its scheduler token).",
+    ],
     "0.66.0": [
         "New: device-enrollment primitives on the importable library -- a hosted Data App can now mint "
         "per-device credentials in-process (no CLI subprocess, no master token on the device). "

src/keboola_agent_cli/client.py

@@ -174,6 +174,7 @@ def __init__(self, stack_url: str, token: str) -> None:
         self._queue_client: httpx.Client | None = None
         self._query_client: httpx.Client | None = None
         self._encrypt_client: httpx.Client | None = None
+        self._scheduler_client: httpx.Client | None = None
         # Lazily built on first Data Streams call (per-device OTLP sources); the
         # Stream control plane is a sibling host reachable from this stack+token.
         self._stream_client: StreamClient | None = None
@@ -196,6 +197,10 @@ def _query_base_url(self) -> str:
     def _encrypt_base_url(self) -> str:
         return self._derive_service_url(self._stack_url, "encryption")
 
+    @property
+    def _scheduler_base_url(self) -> str:
+        return self._derive_service_url(self._stack_url, "scheduler")
+
     def close(self) -> None:
         """Close the underlying HTTP clients."""
         super().close()
@@ -205,6 +210,8 @@ def close(self) -> None:
             self._query_client.close()
         if self._encrypt_client is not None:
             self._encrypt_client.close()
+        if self._scheduler_client is not None:
+            self._scheduler_client.close()
         if self._stream_client is not None:
             self._stream_client.close()
 
@@ -264,6 +271,69 @@ def _encrypt_request(self, method: str, path: str, **kwargs: Any) -> httpx.Respo
             method, path, client=client, base_url=self._encrypt_base_url, **kwargs
         )
 
+    def _scheduler_request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
+        """Execute a Scheduler Service API request with retry."""
+        client = self._get_or_create_sub_client("_scheduler_client", self._scheduler_base_url)
+        return self._do_request(
+            method, path, client=client, base_url=self._scheduler_base_url, **kwargs
+        )
+
+    def activate_schedule(
+        self,
+        configuration_id: str,
+        configuration_version_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Activate a keboola.scheduler configuration in the Scheduler Service.
+
+        Creating a keboola.scheduler *configuration* via the Storage API only
+        stores the cron/target JSON -- it does NOT register the schedule with
+        the Scheduler microservice, so the schedule never fires. Activation is
+        the separate step that registers the schedule (and mints its scheduler
+        token) so it actually runs on the cron cadence.
+
+        Args:
+            configuration_id: The keboola.scheduler configuration ID to activate.
+            configuration_version_id: Optional specific configuration version to
+                pin. Defaults to the latest version when omitted.
+
+        Returns:
+            The activated schedule record, including its ``id`` (schedule ID).
+        """
+        body: dict[str, Any] = {"configurationId": str(configuration_id)}
+        if configuration_version_id is not None:
+            body["configurationVersionId"] = str(configuration_version_id)
+        response = self._scheduler_request("POST", "/schedules", json=body)
+        return response.json()
+
+    def list_activated_schedules(self, configuration_id: str | None = None) -> list[dict[str, Any]]:
+        """List schedules activated in the Scheduler Service for this project.
+
+        Args:
+            configuration_id: Optional keboola.scheduler configuration ID to
+                filter by (client-side); when omitted, returns every activated
+                schedule the token can see.
+
+        Returns:
+            List of activated schedule records (each with ``id`` and
+            ``configurationId``).
+        """
+        response = self._scheduler_request("GET", "/schedules")
+        payload = response.json()
+        schedules = payload if isinstance(payload, list) else payload.get("schedules", [])
+        if configuration_id is None:
+            return schedules
+        return [s for s in schedules if str(s.get("configurationId", "")) == str(configuration_id)]
+
+    def deactivate_schedule(self, schedule_id: str) -> None:
+        """Deactivate (delete) an activated schedule in the Scheduler Service.
+
+        Removes the Scheduler Service registration and its scheduler token so a
+        deleted keboola.scheduler config does not leave an orphaned activation
+        that keeps firing (or errors) against a config that no longer exists.
+        """
+        safe_id = quote(str(schedule_id), safe="")
+        self._scheduler_request("DELETE", f"/schedules/{safe_id}")
+
     def encrypt_values(
         self,
         project_id: int,

src/keboola_agent_cli/commands/flow.py

@@ -829,6 +829,16 @@ def flow_schedule(
             f"  Scheduler config: {escape(result.get('schedule_name', ''))} "
             f"[dim](ID: {escape(result.get('schedule_id', ''))})[/dim]"
         )
+        if result.get("activated"):
+            formatter.console.print(
+                "  [green]Activated[/green] in Scheduler Service "
+                f"[dim](schedule ID: {escape(result.get('activated_schedule_id', ''))})[/dim]"
+            )
+        else:
+            formatter.console.print(
+                "  [yellow]Not activated[/yellow] — dev-branch schedules activate on deploy "
+                "to production."
+            )
 
 
 # ---------------------------------------------------------------------------

src/keboola_agent_cli/services/flow_service.py

@@ -689,19 +689,41 @@ def set_flow_schedule(
                     branch_id=effective_branch,
                 )
                 status = "created"
+
+            scheduler_config_id = str(result.get("id", existing_id or ""))
+
+            # Activate the scheduler config in the Scheduler Service. Creating
+            # the keboola.scheduler config above only stores the cron JSON in
+            # Storage -- without this step the Scheduler microservice never
+            # learns about it and the schedule never fires, even with
+            # state="enabled". Activation is only meaningful on production;
+            # dev-branch schedules are activated when the branch is deployed.
+            activated = False
+            activated_schedule_id: str | None = None
+            if effective_branch is None:
+                schedule = client.activate_schedule(
+                    scheduler_config_id,
+                    configuration_version_id=str(result["version"])
+                    if result.get("version") is not None
+                    else None,
+                )
+                activated = True
+                activated_schedule_id = str(schedule.get("id", ""))
         finally:
             client.close()
 
         return {
             "status": status,
             "project_alias": alias,
-            "schedule_id": str(result.get("id", existing_id or "")),
+            "schedule_id": scheduler_config_id,
             "schedule_name": schedule_name,
             "component_id": FLOW_COMPONENT_ID,
             "config_id": config_id,
             "cron_tab": cron_tab,
             "timezone": timezone,
             "state": "enabled" if enabled else "disabled",
+            "activated": activated,
+            "activated_schedule_id": activated_schedule_id,
             "branch_id": effective_branch,
         }
 
@@ -731,7 +753,23 @@ def remove_flow_schedule(
                 else:
                     raise
 
+            # On production, pull the Scheduler Service registrations once so we
+            # can deactivate them alongside their config. Deleting only the
+            # Storage config would orphan the activation (and its scheduler
+            # token), leaving the Scheduler Service firing against a config that
+            # no longer exists.
+            activated_by_config: dict[str, list[str]] = {}
+            if effective_branch is None:
+                try:
+                    for active in client.list_activated_schedules():
+                        cfg_id = str(active.get("configurationId", ""))
+                        activated_by_config.setdefault(cfg_id, []).append(str(active.get("id", "")))
+                except KeboolaApiError as exc:
+                    if exc.error_code != ErrorCode.NOT_FOUND:
+                        raise
+
             deleted: list[str] = []
+            deactivated: list[str] = []
             errors: list[str] = []
             for sched in all_sched:
                 body = _parse_configuration(sched.get("configuration"))
@@ -740,6 +778,12 @@ def remove_flow_schedule(
                     target.get("configurationId", "")
                 ) == str(config_id):
                     sched_id = str(sched.get("id", ""))
+                    for schedule_id in activated_by_config.get(sched_id, []):
+                        try:
+                            client.deactivate_schedule(schedule_id)
+                            deactivated.append(schedule_id)
+                        except KeboolaApiError as exc:
+                            errors.append(f"{schedule_id}: {exc.message}")
                     try:
                         client.delete_config(
                             SCHEDULER_COMPONENT_ID, sched_id, branch_id=effective_branch
@@ -765,5 +809,6 @@ def remove_flow_schedule(
             "config_id": config_id,
             "deleted_schedule_ids": deleted,
             "deleted_count": len(deleted),
+            "deactivated_schedule_ids": deactivated,
             "branch_id": effective_branch,
         }

tests/test_client.py

@@ -3702,3 +3702,63 @@ def test_rejects_url_without_host(self) -> None:
 
         with pytest.raises(KeboolaApiError):
             _assert_safe_download_url("file:///etc/passwd")
+
+
+_SCHED_TOKEN = "901-55555-fakeTestTokenDoNotUseXXXXXXXX"
+
+
+class TestSchedulerService:
+    """Scheduler Service activation/deactivation (sibling 'scheduler.' host)."""
+
+    def test_activate_schedule_posts_configuration_id(self, httpx_mock) -> None:
+        httpx_mock.add_response(
+            url="https://scheduler.keboola.com/schedules",
+            json={"id": "sched-9", "configurationId": "77"},
+            status_code=201,
+        )
+        client = KeboolaClient(stack_url="https://connection.keboola.com", token=_SCHED_TOKEN)
+        result = client.activate_schedule("77", configuration_version_id="3")
+        assert result["id"] == "sched-9"
+        request = httpx_mock.get_request()
+        assert request.method == "POST"
+        assert json.loads(request.content) == {
+            "configurationId": "77",
+            "configurationVersionId": "3",
+        }
+        assert request.headers["x-storageapi-token"] == _SCHED_TOKEN
+        client.close()
+
+    def test_activate_schedule_omits_version_when_absent(self, httpx_mock) -> None:
+        httpx_mock.add_response(
+            url="https://scheduler.keboola.com/schedules",
+            json={"id": "sched-9"},
+        )
+        client = KeboolaClient(stack_url="https://connection.keboola.com", token=_SCHED_TOKEN)
+        client.activate_schedule("77")
+        request = httpx_mock.get_request()
+        assert json.loads(request.content) == {"configurationId": "77"}
+        client.close()
+
+    def test_list_activated_schedules_filters_by_configuration_id(self, httpx_mock) -> None:
+        httpx_mock.add_response(
+            url="https://scheduler.keboola.com/schedules",
+            json=[
+                {"id": "sched-9", "configurationId": "77"},
+                {"id": "sched-other", "configurationId": "88"},
+            ],
+        )
+        client = KeboolaClient(stack_url="https://connection.keboola.com", token=_SCHED_TOKEN)
+        result = client.list_activated_schedules(configuration_id="77")
+        assert [s["id"] for s in result] == ["sched-9"]
+        client.close()
+
+    def test_deactivate_schedule_deletes_by_id(self, httpx_mock) -> None:
+        httpx_mock.add_response(
+            url="https://scheduler.keboola.com/schedules/sched-9",
+            status_code=204,
+        )
+        client = KeboolaClient(stack_url="https://connection.keboola.com", token=_SCHED_TOKEN)
+        client.deactivate_schedule("sched-9")
+        request = httpx_mock.get_request()
+        assert request.method == "DELETE"
+        client.close()