From e0bfbebe9fb3c7f8c8ee110b950e7a1bba222556 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 9 Jul 2026 11:26:17 +0000 Subject: [PATCH] fix: activate flow schedules in the Scheduler Service (not just write the config) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Zora Jelínková --- .claude-plugin/marketplace.json | 2 +- plugins/kbagent/.claude-plugin/plugin.json | 2 +- .../kbagent/references/commands-reference.md | 4 +- .../skills/kbagent/references/gotchas.md | 20 ++++++ pyproject.toml | 2 +- src/keboola_agent_cli/changelog.py | 9 +++ src/keboola_agent_cli/client.py | 70 +++++++++++++++++++ src/keboola_agent_cli/commands/flow.py | 10 +++ .../services/flow_service.py | 47 ++++++++++++- tests/test_client.py | 60 ++++++++++++++++ tests/test_flow_service.py | 55 ++++++++++++++- uv.lock | 2 +- 12 files changed, 275 insertions(+), 8 deletions(-) diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 47d2dc2b..ed1ee38e 100644 --- a/.claude-plugin/marketplace.json +++ b/.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" diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index 45549029..c63d6a31 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/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", diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 35ade439..680b9e60 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/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 diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index b19d7a7f..76c68188 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 5117765e..0b0b0940 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 9f14f21b..07d74839 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/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). " diff --git a/src/keboola_agent_cli/client.py b/src/keboola_agent_cli/client.py index b6d7e7ba..8f0e3543 100644 --- a/src/keboola_agent_cli/client.py +++ b/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, diff --git a/src/keboola_agent_cli/commands/flow.py b/src/keboola_agent_cli/commands/flow.py index 052d5a34..753a5e45 100644 --- a/src/keboola_agent_cli/commands/flow.py +++ b/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." + ) # --------------------------------------------------------------------------- diff --git a/src/keboola_agent_cli/services/flow_service.py b/src/keboola_agent_cli/services/flow_service.py index d6354551..ab4d32c5 100644 --- a/src/keboola_agent_cli/services/flow_service.py +++ b/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, } diff --git a/tests/test_client.py b/tests/test_client.py index 7547ea39..678d91b5 100644 --- a/tests/test_client.py +++ b/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() diff --git a/tests/test_flow_service.py b/tests/test_flow_service.py index 4a54a79c..e5b084c5 100644 --- a/tests/test_flow_service.py +++ b/tests/test_flow_service.py @@ -396,7 +396,8 @@ def test_set_flow_schedule_targets_keboola_flow(): client = MagicMock() client.get_config_detail.return_value = {"name": "CF"} client.list_component_configs.return_value = [] - client.create_config.return_value = {"id": "77"} + client.create_config.return_value = {"id": "77", "version": 1} + client.activate_schedule.return_value = {"id": "sched-9"} svc = _make_flow_service(client) result = svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *") # scheduler config created with target.componentId == keboola.flow @@ -405,6 +406,34 @@ def test_set_flow_schedule_targets_keboola_flow(): assert result["component_id"] == "keboola.flow" +def test_set_flow_schedule_activates_in_scheduler_service(): + client = MagicMock() + client.get_config_detail.return_value = {"name": "CF"} + client.list_component_configs.return_value = [] + client.create_config.return_value = {"id": "77", "version": 3} + client.activate_schedule.return_value = {"id": "sched-9"} + svc = _make_flow_service(client) + result = svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *") + # The keboola.scheduler config is activated in the Scheduler Service so it + # actually fires -- creating the config alone would never run. + client.activate_schedule.assert_called_once_with("77", configuration_version_id="3") + assert result["activated"] is True + assert result["activated_schedule_id"] == "sched-9" + + +def test_set_flow_schedule_skips_activation_on_dev_branch(): + client = MagicMock() + client.get_config_detail.return_value = {"name": "CF"} + client.list_component_configs.return_value = [] + client.create_config.return_value = {"id": "77", "version": 1} + svc = _make_flow_service(client) + result = svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *", branch_id=123) + # Dev-branch schedules are activated on deploy to production, not here. + client.activate_schedule.assert_not_called() + assert result["activated"] is False + assert result["activated_schedule_id"] is None + + def test_remove_flow_schedule_filters_keboola_flow(): client = MagicMock() client.list_component_configs.return_value = [ @@ -415,7 +444,31 @@ def test_remove_flow_schedule_filters_keboola_flow(): }, } ] + client.list_activated_schedules.return_value = [] svc = _make_flow_service(client) result = svc.remove_flow_schedule(alias="prod", config_id="5") assert result["deleted_count"] == 1 assert result["component_id"] == "keboola.flow" + + +def test_remove_flow_schedule_deactivates_scheduler_service(): + client = MagicMock() + client.list_component_configs.return_value = [ + { + "id": "77", + "configuration": { + "target": {"componentId": "keboola.flow", "configurationId": "5"}, + }, + } + ] + # An activation exists for the keboola.scheduler config being deleted. + client.list_activated_schedules.return_value = [ + {"id": "sched-9", "configurationId": "77"}, + {"id": "sched-other", "configurationId": "88"}, + ] + svc = _make_flow_service(client) + result = svc.remove_flow_schedule(alias="prod", config_id="5") + # Only the matching activation is removed, before the config itself. + client.deactivate_schedule.assert_called_once_with("sched-9") + assert result["deactivated_schedule_ids"] == ["sched-9"] + assert result["deleted_count"] == 1 diff --git a/uv.lock b/uv.lock index 9e5ad008..56711c64 100644 --- a/uv.lock +++ b/uv.lock @@ -590,7 +590,7 @@ wheels = [ [[package]] name = "keboola-cli" -version = "0.66.0" +version = "0.66.1" source = { editable = "." } dependencies = [ { name = "croniter" },