diff --git a/CLAUDE.md b/CLAUDE.md index 376610b79..33429c2b3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1015,6 +1015,27 @@ kbagent flow update --project NAME --flow-id ID [--name N] [--description D] [-- kbagent flow delete --project NAME --flow-id ID [--branch ID] [--yes] kbagent flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--timezone TZ] [--disabled] [--branch ID] kbagent flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes] +kbagent flow run --project NAME --flow-id ID [--from-phase PHASE_ID | --only-task TASK_ID ...] [--dry-run] [--wait] [--timeout N] [--branch ID] [--poll-strategy exponential|fixed] [--log-tail-lines N] +# flow run (#725): run a flow, optionally only PART of it. With neither selector it is the +# ordinary full run (same as `job run --component-id keboola.flow`). With one, kbagent sends +# `onlyFlowTaskIds` on POST /jobs -- a documented Queue API field the CLI never set, which is +# why partial runs were UI-only. --from-phase expands a phase id into its tasks plus every task +# in every phase reachable via next[].goto (disabled tasks in scope are skipped silently); +# --only-task takes explicit ids (an unknown OR disabled id is a hard error -- the caller named +# it). The two are mutually exclusive. --dry-run prints the resolved selection and creates NO +# job; it REQUIRES a selector (a full run has nothing to preview). +# CRITICAL SEMANTICS -- a selected run IGNORES the flow's phase conditions. The Queue daemon +# linearizes the phase graph into one sequential chain of synthetic unconditional `selected-run` +# transitions, phases kept only to route to a selected task run EMPTY, a failing task does NOT +# stop the chain, and the flow's final status is that of the LAST phase in the chain (so a run +# whose last phase succeeded reports `success` even though an earlier selected task failed). +# Unselected variable tasks do not run, so their flow variables are not merged. Use this to +# RE-RUN part of a flow -- never to verify that new next[].condition logic evaluates correctly; +# no API does the latter without a full run. Every payload carries `conditions_evaluated: false`. +# A rejected selector (unknown phase, unknown/disabled task, empty selection) exits 2, and an +# unreachable --from-phase is refused locally before a job is created. Result JSON gains +# `flow_task_selection` (ids + names + phases) so a caller keeping only the run payload still +# knows what ran. Over `serve`: POST /flows/{project}/{config_id}/run. kbagent flow triggers --project NAME --flow-id ID [--branch ID] # flow triggers (#714): the honest answer to "what starts this flow automatically". A flow has at # least three trigger mechanisms and kbagent used to see ONE: `schedule list` / `search` only ever diff --git a/docs/web-server-endpoints.md b/docs/web-server-endpoints.md index a7b50443b..b2a803135 100644 --- a/docs/web-server-endpoints.md +++ b/docs/web-server-endpoints.md @@ -9,7 +9,7 @@ auth, and the concepts behind these routes live in [`web-server.md`](web-server.md); a running server serves the same spec interactively at `/docs` (Swagger) and `/openapi.json`. -**235 operations** across **205 paths** and **30 routers**. +**236 operations** across **206 paths** and **30 routers**. Paths are shown as the server registers them. Reaching them through the Node BFF (or single-process `--ui` mode) prefixes every path with `/api`. @@ -251,7 +251,7 @@ Run components, inspect job history, terminate running jobs. Mirrors `kbagent jo | `POST` | `/jobs/{project}/terminate` | Terminate running jobs | | `GET` | `/jobs/{project}/{job_id}/stream` | Stream job status and logs (SSE) | -### `flows` (12 operations) +### `flows` (13 operations) Orchestrator and Flow CRUD, scheduling, run history. Mirrors `kbagent flow *`. @@ -267,6 +267,7 @@ Orchestrator and Flow CRUD, scheduling, run history. Mirrors `kbagent flow *`. | `POST` | `/flows/{project}` | Create a new flow | | `GET` | `/flows/{project}/{config_id}/schedules` | List schedules for a flow | | `GET` | `/flows/{project}/{config_id}/triggers` | List every trigger kbagent can see | +| `POST` | `/flows/{project}/{config_id}/run` | Run a flow, optionally only part of it | | `POST` | `/flows/{project}/{config_id}/schedule` | Set a cron schedule on a flow | | `DELETE` | `/flows/{project}/{config_id}/schedule` | Remove a flow schedule | diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md index b302c00fd..8f811ac6d 100644 --- a/plugins/kbagent/agents/keboola-expert.md +++ b/plugins/kbagent/agents/keboola-expert.md @@ -100,6 +100,7 @@ been retired, so its absence is NOT a promise (see §1 Rule 6). |---|---|---|---| | Author / edit a conditional flow (keboola.flow) | `kbagent flow validate --file @flow.yaml --project P` (fetches the live schema; loop until clean) then `kbagent flow new` / `flow update --file` | fetch `flow detail`, merge phases/tasks locally, re-validate, push | integer ids (ids are STRINGS); `dependsOn` (use `next[].goto` + conditions); `--component-id` or `keboola.orchestrator` (both dropped 0.57.0); `flow schema --full` without `--project` | | Schedule flow | `kbagent flow schedule --cron ... [--timezone]` -- confirm `activated: true` (older versions wrote a config whose cron NEVER fired) | -- | raw REST to `/storage/configurations/keboola.scheduler` | +| Re-run part of a flow / run a flow from a phase onward | `kbagent flow run --project P --flow-id ID --from-phase PHASE --dry-run` to preview, then without `--dry-run` (vNEXT+) | `--only-task ID` for an explicit task allowlist; a full `kbagent flow run` when the conditions themselves matter | claiming a partial run VERIFIES new `next[].condition` logic -- it does NOT, conditions are ignored; reading a partial run's `success` as "every selected task passed" (the chain does not stop on failure and the final status is the LAST phase's); `job run --component-id keboola.flow` when only one phase needs re-running | | What starts this flow automatically / is this flow scheduled | `kbagent flow triggers --project P --flow-id ID` (0.92.0+) -- cron schedules AND table triggers in one call | `kbagent schedule list` for the cron half only | concluding "this flow has no trigger" from an empty result -- `cross_project_triggers_checked` is `false`, so empty means "no trigger that kbagent CHECKED"; reading `last_run: null` as "disabled" (it means never fired); expecting table triggers on a dev branch (the Storage route is production-only) | | Who gets notified when a flow fails | `kbagent notification list [--component-id keboola.flow]` (0.86.0+) -- recipients live in a separate service, NOT the flow config | `flow detail` for in-flow `notification` TASKS | reading "nobody is notified" off a flow config, or off a filtered run (check `project_wide_excluded`); camelCase event names (they are kebab-case) | | Add / remove / re-point a notification recipient | `kbagent notification create --event job-failed --channel email\|webhook --address A [--config-id K] [--branch ID]` / `notification delete --subscription-id ID` / `notification replace-recipient --subscription-id ID --address NEW` (0.91.0+) | -- | caching the old id after a replace (it is delete+recreate -- a NEW `subscription_id` is always minted; a failed delete leaves a duplicate as `old_deleted: false`); omitting `--branch` and expecting the UI's behavior (no `--branch` = NO `branch.id` filter = fires on every branch) | @@ -247,6 +248,17 @@ its absence is NOT a promise the entry is version-independent (see §1 Rule 6). as needing manual scheduling because of exactly this gap (#714). For the same reason `job detail` on a flow job carries `trigger_hint`: a run with no matching cron schedule is NOT proof of a manual run. +- **A partial `flow run` IGNORES the flow's conditions (vNEXT+)**: + `--from-phase` / `--only-task` send the Queue API's `onlyFlowTaskIds`, and the + daemon then linearizes the phase graph into an UNCONDITIONAL chain: no + `next[].condition` is evaluated, routing-only phases run empty, a failing + task does not stop the run, and the flow's final status is the LAST phase's + -- so a partial run can report `success` with a failed task inside it. Read + the per-task results, never the job status alone. It is the right tool to + RE-RUN part of a flow cheaply; it is the wrong tool to prove new condition + logic works (nothing short of a full run does that -- use a dev branch). + Preview any selection with `--dry-run` first: `--from-phase` expands to every + downstream phase's tasks, which is usually more than the caller pictured. - **`flow schedule` activates on the Scheduler Service**: older versions only wrote the `keboola.scheduler` config -- it showed `enabled` but the cron NEVER fired. Confirm `activated: true`; `activated: false` + warning diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index 5b70f8ad0..c346c2010 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -225,6 +225,7 @@ When working inside a git repository or project directory, run `kbagent init` (o | Bind a cron schedule to a flow (upsert: creates or updates) | `kbagent flow schedule --project PROJECT --flow-id FLOW-ID --cron CRON` | | Remove all schedules bound to a flow (deletes keboola.scheduler configs) | `kbagent flow schedule-remove --project PROJECT --flow-id FLOW-ID` | | Show every trigger kbagent can see for a flow (cron + table triggers) | `kbagent flow triggers --project PROJECT --flow-id FLOW-ID` | +| Run a flow, optionally only from a given phase onward | `kbagent flow run --project PROJECT --flow-id FLOW-ID` | | List cron schedules (keboola.scheduler configs) across projects | `kbagent schedule list` | | Show full detail for a single cron schedule | `kbagent schedule detail --project PROJECT --schedule-id SCHEDULE-ID` | | Audit schedules by cron window or job-freshness | `kbagent schedule find` | diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 83281d11d..8c0ed3530 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -308,6 +308,7 @@ Requires the project to be added with its **master ('owner') Storage API token** - `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. Since v0.66.1 the config is also activated on the Scheduler Service so the trigger fires; an activation failure keeps the config written and surfaces a warning (`activated: false`, exit 0) - `flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes]` -- remove all cron schedules attached to a flow (since v0.66.1 also deregisters them from the Scheduler Service); idempotent +- `flow run --project NAME --flow-id ID [--from-phase PHASE_ID | --only-task TASK_ID ...] [--dry-run] [--wait] [--timeout N] [--branch ID] [--poll-strategy exponential|fixed] [--log-tail-lines N]` -- run a flow, optionally only **part** of it (since vNEXT, #725). With neither selector this is the ordinary full run, identical to `job run --component-id keboola.flow`. With one, kbagent sends `onlyFlowTaskIds` on `POST /jobs` -- a documented Queue API field the CLI never set, which is why a partial run used to be UI-only. `--from-phase` expands a phase id into its own tasks **plus every task in every phase reachable via `next[].goto`** (disabled tasks in scope are skipped silently); `--only-task` takes explicit ids, where an unknown **or disabled** id is a hard error -- the caller named it. The two are mutually exclusive. `--dry-run` prints the resolved selection and creates **no** job; it requires a selector, since a full run has nothing to preview. **A selected run IGNORES the flow's phase conditions** -- see [gotchas.md](gotchas.md); every payload carries `conditions_evaluated: false`. A rejected selector exits 2, and an unreachable `--from-phase` is refused locally before any job is created. The run result gains `flow_task_selection` (ids + names + phases) so a caller keeping only the job payload still knows what ran - `flow triggers --project NAME --flow-id ID [--branch ID]` -- **the honest answer to "what starts this flow automatically"** (since 0.92.0, #714). Returns `cron_schedules` (the `keboola.scheduler` configs `schedule list` already saw) **and** `table_triggers` (`GET /v2/storage/triggers` -- a separate Storage resource, NOT a component config, which is why no kbagent command saw them before). Cross-project triggers (a trigger-queue app config in ANOTHER project) are **not** covered: the result carries `cross_project_triggers_checked: false`, deliberately not an empty list -- an empty list reads as "checked, found none", the exact false negative this command exists to prevent. **Check that flag before concluding a flow has no trigger.** Table triggers are **production-only** (the Storage route is declared `isAvailableInBranch: false`), so `--branch` narrows the cron half only and `table_triggers_branch_scoped` is always `false`. A `last_run` of `null` means the trigger exists but has never fired -- not that it is disabled ## Schedule Discovery & Audit (Fleet-Wide) diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index fd33996b9..7ae723bc9 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -4981,3 +4981,49 @@ single-project command: project is not the first row of `project list`. - The fix also covers `kbagent serve`: the `/kai/*` routes, `POST /documentation/query`, and `GET /components?query=...` resolved the project the same wrong way. + + +## A partial `flow run` does not evaluate the flow's conditions + +*(since vNEXT)* `kbagent flow run --from-phase` / `--only-task` sends the Queue +API's `onlyFlowTaskIds`, and the job-queue daemon treats that as a **task +selection**, not as "start the flow here". Per +`keboola/job-queue-daemon` `docs/flow-documentation.md`, such a run: + +1. runs **only** the selected tasks -- every other task, notification and + variable task included, is left out entirely and does not appear in the + result; +2. **does not evaluate any condition.** The phase graph is linearized into one + sequential chain of synthetic unconditional transitions with id + `selected-run` (visible in the result's `phases[].conditions[]`); +3. traverses phases kept only to route to a selected task as **empty phases** -- + they run an empty phase container and report `success` with no tasks; +4. **does not stop on a failing task.** The transitions are unconditional, so + the chain continues through a failed or warning phase; +5. reports as the flow's final status the status of the **last phase in the + chain**. Combined with (4), a selected run whose last phase succeeded + reports `success` even though an earlier selected task **failed** -- the + failure is visible in the task and phase entries, not in the job status. + **Never read a partial run's `success` as "everything in it passed"**; read + the per-task results; +6. does not merge flow variables defined by unselected variable tasks -- the + selected tasks fall back to their component variable configuration. + +The consequence worth internalising: **a partial run cannot verify that new +`next[].condition` logic evaluates correctly.** That was the motivation in +issue #725, and no API satisfies it short of a full run -- the conditions are +precisely the thing a selected run throws away. Use `flow run` to *re-run* part +of a flow cheaply; to exercise a new condition, run the whole flow (a dev +branch keeps that cheap). Every payload from the command carries +`conditions_evaluated: false` so an agent cannot conclude otherwise from the +JSON alone. + +kbagent pre-empts the daemon's own user-errors locally, before a job exists: +an unknown phase id (the message lists the valid ones), a `--from-phase` that +is not reachable from the flow's entry phase, an unknown or **disabled** +`--only-task` id, and a selection that resolves to nothing all exit **2** +without creating a job. A disabled task inside a `--from-phase` scope is +different: it is skipped silently and reported under +`skipped_disabled_task_ids`, because the caller selected a phase, not that +task. + diff --git a/src/keboola_agent_cli/client/queue.py b/src/keboola_agent_cli/client/queue.py index 1099c6d1e..0b4f01ee7 100644 --- a/src/keboola_agent_cli/client/queue.py +++ b/src/keboola_agent_cli/client/queue.py @@ -126,6 +126,7 @@ def create_job( mode: str = "run", branch_id: int | None = None, variable_values_id: str | None = None, + only_flow_task_ids: list[str] | None = None, ) -> dict[str, Any]: """Create and run a Queue API job. @@ -142,6 +143,16 @@ def create_job( ``keboola.variables`` config. When set, the Queue API binds the row's values to the job's `{{ variable }}` placeholders. Omit for configurations that have no linked variables. + only_flow_task_ids: Optional ``keboola.flow`` task-id allowlist + (issue #725). Only meaningful for a flow job; the Queue API + then runs ONLY those tasks. Note the daemon **linearizes the + phase graph and ignores every configured condition** in such + a run -- a selected run is a partial re-execution, not a + rehearsal of the flow's ``next[].condition`` logic. Must be + non-empty when set (the API rejects an empty array), and the + deprecated ``onlyOrchestrationTaskIds`` alias is deliberately + not sent: the API rejects the two together and mirrors this + one back under both keys anyway. Returns: Job dict from the Queue API. @@ -159,6 +170,8 @@ def create_job( body["configRowIds"] = config_row_ids if variable_values_id: body["variableValuesId"] = variable_values_id + if only_flow_task_ids: + body["onlyFlowTaskIds"] = only_flow_task_ids response = self._queue_request("POST", "/jobs", json=body) return response.json() diff --git a/src/keboola_agent_cli/commands/_flow_run.py b/src/keboola_agent_cli/commands/_flow_run.py new file mode 100644 index 000000000..c600437fa --- /dev/null +++ b/src/keboola_agent_cli/commands/_flow_run.py @@ -0,0 +1,282 @@ +"""The ``kbagent flow run`` command (issue #725). + +Running a flow used to mean ``job run --component-id keboola.flow``, which has +no phase or task awareness: every run starts at phase 1 and re-executes the +whole graph. Re-testing one downstream phase of a 16-phase production flow +therefore meant re-running ~50 unrelated tasks, or running the components +individually and bypassing the flow engine entirely. + +The Queue API has supported a task allowlist all along -- ``onlyFlowTaskIds`` +on ``POST /jobs`` (keboola/job-queue ``apps/public-api/docs/swagger.yaml``) -- +and mirrors it back on every job detail. kbagent simply never sent it. This +command sends it, and resolves a *phase* into the task ids for callers who +think in phases rather than in task ids. + +WHAT A SELECTED RUN IS AND IS NOT (keboola/job-queue-daemon +``docs/flow-documentation.md``): the daemon runs only the selected tasks and +**ignores every configured condition** -- the phase graph is linearized into a +single sequential chain of synthetic unconditional ``selected-run`` +transitions. Phases kept only to route to a selected task run empty. A failing +task does not stop the chain. So this is "re-run this part of the flow", NOT +"rehearse the flow's ``next[].condition`` logic": there is no API that does the +latter without a full run, and pretending otherwise would be worse than the +gap it fills. The command says so in ``--dry-run`` and in JSON +(``conditions_evaluated: false``). + +Lives in a private module because ``commands/flow.py`` is at the 800-code-line +commands soft ceiling (CONTRIBUTING.md file-size budgets). Mounted flat onto +``flow_app`` via :func:`register`, so the permission key stays ``flow.run``. +""" + +from __future__ import annotations + +from typing import Any + +import typer +from rich.markup import escape +from rich.table import Table + +from ..constants import ( + DEFAULT_JOB_RUN_TIMEOUT, + DEFAULT_LOG_TAIL_LINES, + DEFAULT_POLL_STRATEGY, + MAX_LOG_TAIL_LINES, +) +from ..errors import ConfigError, ErrorCode, KeboolaApiError +from ._helpers import ( + get_formatter, + get_service, + map_error_to_exit_code, + resolve_branch, +) + +CONDITIONS_NOTE = ( + "A selected run IGNORES the flow's phase conditions: the graph is " + "linearized into an unconditional chain, and a failing task does not stop " + "it. Use it to re-run part of a flow, not to verify next[].condition logic." +) + + +def _format_selection(formatter: Any, selection: dict[str, Any]) -> None: + """Human-mode preview of what a partial run resolved to.""" + task_ids = selection.get("task_ids") or [] + formatter.console.print( + f"\n[bold]Partial run of flow {escape(str(selection.get('config_id', '')))}[/bold]" + f" [dim]({escape(str(selection.get('project_alias', '')))})[/dim]" + ) + if selection.get("from_phase"): + formatter.console.print( + f"[dim]From phase: {escape(str(selection['from_phase']))} " + f"(plus every phase reachable from it)[/dim]" + ) + + table = Table(title=f"Selected tasks ({len(task_ids)})") + table.add_column("Task ID") + table.add_column("Name") + table.add_column("Phase") + phase_names = {p["id"]: p.get("name", "") for p in selection.get("selected_phases") or []} + for task in selection.get("selected_tasks") or []: + phase_id = str(task.get("phase", "")) + phase_label = phase_id + if phase_names.get(phase_id): + phase_label = f"{phase_id} ({phase_names[phase_id]})" + table.add_row( + escape(str(task.get("id", ""))), + escape(str(task.get("name", ""))), + escape(phase_label), + ) + formatter.console.print(table) + + skipped = selection.get("skipped_disabled_task_ids") or [] + if skipped: + formatter.console.print( + f"[dim]Skipped {len(skipped)} disabled task(s): {escape(', '.join(skipped))}[/dim]" + ) + formatter.console.print(f"\n[yellow]{escape(CONDITIONS_NOTE)}[/yellow]") + + +def _print_run_result( + formatter: Any, + result: dict[str, Any], + project: str, + selection: dict[str, Any] | None, + wait: bool, +) -> None: + job_id = result.get("id", "?") + status = result.get("status", "unknown") + if selection: + formatter.console.print( + f"[dim]Partial run: {len(selection['task_ids'])} task(s) selected. " + f"{CONDITIONS_NOTE}[/dim]" + ) + if status == "success": + formatter.console.print(f"[bold green]Flow job {job_id}:[/bold green] {status}") + elif status in ("warning", "terminated"): + formatter.console.print(f"[bold yellow]Flow job {job_id}:[/bold yellow] {status}") + else: + formatter.console.print(f"[bold blue]Flow job {job_id}:[/bold blue] {status}") + if not wait: + formatter.console.print( + " Use --wait to poll until completion, " + f"or: kbagent job detail --project {project} --job-id {job_id}" + ) + + +def register(app: typer.Typer) -> None: + """Mount the ``run`` command onto ``app`` (the ``flow`` Typer group).""" + + @app.command("run") + def flow_run( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + flow_id: str = typer.Option(..., "--flow-id", help="Flow configuration ID"), + from_phase: str | None = typer.Option( + None, + "--from-phase", + help=( + "Run this phase and every phase reachable from it via " + "next[].goto. Disabled tasks in scope are skipped. Mutually " + "exclusive with --only-task." + ), + ), + only_task: list[str] | None = typer.Option( + None, + "--only-task", + help=( + "Explicit task ID to run (repeat for several). An unknown or " + "disabled id is an error, not a silent skip. Mutually " + "exclusive with --from-phase." + ), + ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Resolve and print the task selection without creating a job.", + ), + wait: bool = typer.Option( + False, "--wait", help="Wait for the flow job to finish (poll until terminal state)" + ), + timeout: float = typer.Option( + DEFAULT_JOB_RUN_TIMEOUT, "--timeout", help="Max seconds to wait when --wait is set" + ), + branch: int | None = typer.Option( + None, "--branch", help="Dev branch ID (overrides active branch)" + ), + poll_strategy: str = typer.Option( + DEFAULT_POLL_STRATEGY, + "--poll-strategy", + help="Polling cadence used with --wait: 'exponential' (default) or 'fixed'.", + ), + log_tail_lines: int = typer.Option( + DEFAULT_LOG_TAIL_LINES, + "--log-tail-lines", + help=( + "On a non-success terminal state, surface the last N job events. " + f"Only used with --wait. 0 disables; max {MAX_LOG_TAIL_LINES}." + ), + ), + ) -> None: + """Run a flow, optionally only from a given phase onward. + + With neither --from-phase nor --only-task this is the ordinary full + flow run (identical to `job run --component-id keboola.flow`), with + conditions evaluated as configured. + + With either selector the Queue API runs ONLY the selected tasks and + IGNORES the flow's phase conditions -- the graph is linearized into an + unconditional chain, so a failed task does not halt the run and the + flow's final status is that of the last phase in the chain. Preview a + selection with --dry-run before committing to it. + """ + formatter = get_formatter(ctx) + config_store = ctx.obj["config_store"] + + if from_phase and only_task: + formatter.error( + message=( + "--from-phase and --only-task are mutually exclusive. Pass a " + "phase to expand into its downstream tasks, or an explicit " + "task-id allowlist, but not both." + ), + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) + if dry_run and not (from_phase or only_task): + formatter.error( + message=( + "--dry-run needs --from-phase or --only-task: with no selector " + "there is nothing to resolve, and a full flow run has no preview." + ), + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) + if log_tail_lines < 0 or log_tail_lines > MAX_LOG_TAIL_LINES: + formatter.error( + message=( + f"--log-tail-lines must be between 0 and {MAX_LOG_TAIL_LINES}. " + f"Got {log_tail_lines}." + ), + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) + + _, effective_branch = resolve_branch(config_store, formatter, project, branch) + + selection: dict[str, Any] | None = None + try: + if from_phase or only_task: + selection = get_service(ctx, "flow_service").resolve_flow_task_ids( + alias=project, + config_id=flow_id, + from_phase=from_phase, + only_task_ids=list(only_task) if only_task else None, + branch_id=effective_branch, + ) + if dry_run: + if formatter.json_mode: + formatter.output({**selection, "dry_run": True}) + else: + _format_selection(formatter, selection) + return + + result = get_service(ctx, "job_service").run_job( + alias=project, + component_id="keboola.flow", + config_id=flow_id, + wait=wait, + timeout=timeout, + branch_id=effective_branch, + poll_strategy=poll_strategy, + log_tail_lines=log_tail_lines, + only_flow_task_ids=selection["task_ids"] if selection else None, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error( + message=exc.message, + error_code=exc.error_code, + project=project, + retryable=exc.retryable, + details=exc.details or None, + ) + # A rejected selector is a usage error whichever layer caught it: + # the two checks above exit 2, and so must the service-side ones + # (unknown phase, unknown/disabled task id, empty selection). The + # generic mapping sends INVALID_ARGUMENT to exit 1, which would + # make "you typed a bad phase id" indistinguishable from "the run + # failed". + if exc.error_code == ErrorCode.INVALID_ARGUMENT: + raise typer.Exit(code=2) from None + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if selection: + # Echo the selection onto the job payload so a caller that only + # keeps the run's JSON can still tell WHICH tasks it ran -- the + # Queue API mirrors the ids back, but not their names or phases. + result["flow_task_selection"] = selection + if formatter.json_mode: + formatter.output(result) + else: + _print_run_result(formatter, result, project, selection, wait=wait) diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index 012f64668..2f65d59b6 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -1081,6 +1081,17 @@ surfaces a warning (exit stays 0). Re-run with a capable token to activate. kbagent flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes] + kbagent flow run --project NAME --flow-id ID [--from-phase PHASE_ID | --only-task TASK_ID ...] [--dry-run] [--wait] [--timeout N] [--branch ID] [--poll-strategy exponential|fixed] [--log-tail-lines N] + Run a flow, optionally only part of it. No selector = ordinary full run (identical to + `job run --component-id keboola.flow`). --from-phase expands a phase into its tasks plus every + task in every phase reachable via next[].goto; --only-task takes explicit ids. Mutually + exclusive. --dry-run previews the resolved selection without creating a job (needs a selector). + A SELECTED RUN IGNORES THE FLOW'S PHASE CONDITIONS: the graph is linearized into an + unconditional chain, phases kept only for routing run empty, a failing task does not stop the + run, and the final status is that of the last phase in the chain. It re-runs part of a flow; + it does NOT verify next[].condition logic. Payloads carry conditions_evaluated=false. + Unknown/unreachable phase, unknown or disabled --only-task id, or an empty selection => exit 2. + kbagent flow triggers --project NAME --flow-id ID [--branch ID] # Cron schedules AND table triggers in one call. Cross-project triggers are # NOT checked -- the result says so via cross_project_triggers_checked=false diff --git a/src/keboola_agent_cli/commands/flow.py b/src/keboola_agent_cli/commands/flow.py index 4cec2e130..b13fbf002 100644 --- a/src/keboola_agent_cli/commands/flow.py +++ b/src/keboola_agent_cli/commands/flow.py @@ -1047,6 +1047,8 @@ def flow_schedule_remove( # commands soft ceiling, so `triggers` lives in its own file (CONTRIBUTING.md # file-size budgets). Flat registration keeps the permission key `flow.triggers` # and the `kbagent flow --help` listing unchanged. +from ._flow_run import register as _register_run_command # noqa: E402 from ._flow_triggers import register as _register_trigger_commands # noqa: E402 _register_trigger_commands(flow_app) +_register_run_command(flow_app) diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index adfbc329d..5c59cc6d5 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -349,6 +349,7 @@ "flow.list": "read", "flow.detail": "read", "flow.triggers": "read", + "flow.run": "write", "flow.schema": "read", "flow.examples": "read", "flow.validate": "read", diff --git a/src/keboola_agent_cli/server/routers/flows.py b/src/keboola_agent_cli/server/routers/flows.py index 9a87ca3a7..11fdc1809 100644 --- a/src/keboola_agent_cli/server/routers/flows.py +++ b/src/keboola_agent_cli/server/routers/flows.py @@ -7,6 +7,11 @@ from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel +from ...constants import ( + DEFAULT_JOB_RUN_TIMEOUT, + DEFAULT_LOG_TAIL_LINES, + DEFAULT_POLL_STRATEGY, +) from ...services.flow_service import FLOW_COMPONENT_ID, get_flow_examples from ...services.flow_validation import find_unreachable_phases, validate_conditional_flow from ..dependencies import ServiceRegistry, get_registry @@ -38,6 +43,20 @@ class FlowSchedule(BaseModel): branch_id: int | None = None +class FlowRun(BaseModel): + """Body for a (possibly partial) flow run. + + ``from_phase`` and ``only_task_ids`` are mutually exclusive; with neither, + this is an ordinary full flow run. With either, the Queue API runs ONLY the + selected tasks and IGNORES the flow's phase conditions -- see + ``commands/_flow_run.py`` for why that distinction matters. + """ + + from_phase: str | None = None + only_task_ids: list[str] | None = None + branch_id: int | None = None + + class FlowValidate(BaseModel): phases: list[dict[str, Any]] = [] tasks: list[dict[str, Any]] = [] @@ -209,6 +228,62 @@ def list_triggers( return registry.flow.get_flow_triggers(alias=project, config_id=config_id, branch_id=branch_id) +@router.post("/{project}/{config_id}/run", summary="Run a flow, optionally only part of it") +def run( + project: str, + config_id: str, + body: FlowRun, + dry_run: bool = False, + wait: bool = False, + timeout: float = DEFAULT_JOB_RUN_TIMEOUT, + poll_strategy: str = DEFAULT_POLL_STRATEGY, + log_tail_lines: int = DEFAULT_LOG_TAIL_LINES, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Run a flow, optionally limited to a phase or an explicit task allowlist. + + Mirrors `kbagent flow run`. `dry_run=true` resolves the selection and + returns it without creating a job; it requires a selector, since a full + run has nothing to preview. + """ + if body.from_phase and body.only_task_ids: + raise HTTPException( + status_code=400, + detail="from_phase and only_task_ids are mutually exclusive.", + ) + selection: dict[str, Any] | None = None + if body.from_phase or body.only_task_ids: + selection = registry.flow.resolve_flow_task_ids( + alias=project, + config_id=config_id, + from_phase=body.from_phase, + only_task_ids=body.only_task_ids, + branch_id=body.branch_id, + ) + elif dry_run: + raise HTTPException( + status_code=400, + detail="dry_run requires from_phase or only_task_ids.", + ) + if dry_run and selection is not None: + return {**selection, "dry_run": True} + + result = registry.job.run_job( + alias=project, + component_id=FLOW_COMPONENT_ID, + config_id=config_id, + wait=wait, + timeout=timeout, + branch_id=body.branch_id, + poll_strategy=poll_strategy, + log_tail_lines=log_tail_lines, + only_flow_task_ids=selection["task_ids"] if selection else None, + ) + if selection: + result["flow_task_selection"] = selection + return result + + @router.post("/{project}/{config_id}/schedule", summary="Set a cron schedule on a flow") def set_schedule( project: str, diff --git a/src/keboola_agent_cli/services/flow_service.py b/src/keboola_agent_cli/services/flow_service.py index 3652961bd..834005e1e 100644 --- a/src/keboola_agent_cli/services/flow_service.py +++ b/src/keboola_agent_cli/services/flow_service.py @@ -29,7 +29,11 @@ from ..models import ComponentDetail, ProjectConfig from ..scheduler_client import SchedulerClient from .base import BaseService, ClientFactory, project_error_entry -from .flow_validation import find_unreachable_phases, validate_conditional_flow +from .flow_validation import ( + find_unreachable_phases, + reachable_phases, + validate_conditional_flow, +) logger = logging.getLogger(__name__) @@ -480,6 +484,192 @@ def get_flow_detail( detail["task_count"] = len(tasks) return detail + # ── partial-run task selection (issue #725) ───────────────────── + + def resolve_flow_task_ids( + self, + alias: str, + config_id: str, + from_phase: str | None = None, + only_task_ids: list[str] | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Resolve a ``keboola.flow`` task-id allowlist for a partial run. + + Exactly one selector must be given: + + - ``from_phase`` -- take every enabled task in that phase and in every + phase reachable from it via ``next[].goto``. Disabled tasks are + dropped silently: the Queue API rejects a selection containing one, + and a phase-level selection never named them individually. + - ``only_task_ids`` -- an explicit allowlist. Unknown or disabled ids + are a hard error here, because the caller named them: silently + dropping a task the caller asked for would run something other than + what was requested. + + WHAT A SELECTED RUN IS NOT: the Queue daemon linearizes the phase + graph and runs the selection through synthetic unconditional + transitions, so ``next[].condition`` expressions are NOT evaluated + (keboola/job-queue-daemon ``docs/flow-documentation.md``). This + resolves "run this part of the flow again", never "rehearse the + flow's condition logic". + + Returns: + Dict with ``task_ids`` plus the resolution context (selected + phases, skipped disabled tasks) so the caller can show what a + ``--from-phase`` expanded to before anything runs. + + Raises: + ConfigError: If alias is not found. + KeboolaApiError: On API failure, or INVALID_ARGUMENT when the + selector is unusable (bad phase id, unknown/disabled task id, + empty resolution). + """ + if (from_phase is None) == (only_task_ids is None): + raise KeboolaApiError( + message=( + "Pass exactly one of from_phase / only_task_ids to select " + "the tasks a partial flow run should execute." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + + detail = self.get_flow_detail(alias=alias, config_id=config_id, branch_id=branch_id) + phases: list[dict[str, Any]] = detail.get("phases") or [] + tasks: list[dict[str, Any]] = detail.get("tasks") or [] + + if only_task_ids is not None: + selected, skipped = self._select_named_tasks(tasks, only_task_ids) + selected_phase_ids = [str(t.get("phase")) for t in selected] + else: + selected, skipped, selected_phase_ids = self._select_tasks_from_phase( + phases, tasks, str(from_phase) + ) + + task_ids = [str(t.get("id")) for t in selected] + if not task_ids: + raise KeboolaApiError( + message=( + f"Task selection for flow {config_id} resolved to no runnable tasks" + + ( + f" (every task from phase '{from_phase}' onward is disabled)" + if from_phase + else "" + ) + + ". The Queue API rejects an empty selection." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + + phase_names = {str(p.get("id")): str(p.get("name") or "") for p in phases} + ordered_phase_ids = [pid for pid in dict.fromkeys(selected_phase_ids)] + return { + "project_alias": alias, + "component_id": FLOW_COMPONENT_ID, + "config_id": config_id, + "branch_id": detail.get("branch_id"), + "from_phase": from_phase, + "task_ids": task_ids, + "selected_tasks": [ + { + "id": str(t.get("id")), + "name": str(t.get("name") or ""), + "phase": str(t.get("phase")), + } + for t in selected + ], + "selected_phases": [ + {"id": pid, "name": phase_names.get(pid, "")} for pid in ordered_phase_ids + ], + "skipped_disabled_task_ids": skipped, + "conditions_evaluated": False, + } + + @staticmethod + def _select_named_tasks( + tasks: list[dict[str, Any]], only_task_ids: list[str] + ) -> tuple[list[dict[str, Any]], list[str]]: + """Validate an explicit task-id allowlist against the flow definition. + + Returns the matching task dicts in flow order (NOT in the order the + caller listed them -- the run order is the flow's, so presenting the + caller's order would misrepresent it). ``skipped`` is always empty: + a named task is never silently dropped. + """ + by_id = {str(t.get("id")): t for t in tasks} + wanted = list(dict.fromkeys(str(tid) for tid in only_task_ids)) + + unknown = [tid for tid in wanted if tid not in by_id] + if unknown: + raise KeboolaApiError( + message=( + f"Unknown task id(s): {', '.join(unknown)}. " + "Run `kbagent flow detail` to list the flow's task ids." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + disabled = [tid for tid in wanted if not by_id[tid].get("enabled", True)] + if disabled: + raise KeboolaApiError( + message=( + f"Task id(s) disabled in the flow: {', '.join(disabled)}. " + "The Queue API rejects a selection containing a disabled " + "task -- enable it in the flow or drop it from --only-task." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + + chosen = set(wanted) + return [t for t in tasks if str(t.get("id")) in chosen], [] + + @staticmethod + def _select_tasks_from_phase( + phases: list[dict[str, Any]], + tasks: list[dict[str, Any]], + from_phase: str, + ) -> tuple[list[dict[str, Any]], list[str], list[str]]: + """Expand a phase id into the tasks of it and everything downstream.""" + known_phase_ids = [str(p.get("id")) for p in phases] + if from_phase not in known_phase_ids: + raise KeboolaApiError( + message=( + f"Unknown phase id '{from_phase}'. " + f"Known phases: {', '.join(known_phase_ids) or '(none)'}." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + # The daemon refuses a selection whose phases cannot be reached from + # the flow's entry phase, so catch that here rather than paying for a + # round trip to have the job rejected. + if from_phase in find_unreachable_phases(phases): + raise KeboolaApiError( + message=( + f"Phase '{from_phase}' is not reachable from the flow's entry " + f"phase '{known_phase_ids[0]}', so the Queue API cannot route " + "a run to it." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + + downstream = set(reachable_phases(phases, from_phase)) + in_scope = [t for t in tasks if str(t.get("phase")) in downstream] + selected = [t for t in in_scope if t.get("enabled", True)] + skipped = [str(t.get("id")) for t in in_scope if not t.get("enabled", True)] + selected_phase_ids = [pid for pid in known_phase_ids if pid in downstream] + return selected, skipped, selected_phase_ids + # ── create ────────────────────────────────────────────────────── def create_flow( diff --git a/src/keboola_agent_cli/services/flow_validation.py b/src/keboola_agent_cli/services/flow_validation.py index 81eda845e..09a6e7563 100644 --- a/src/keboola_agent_cli/services/flow_validation.py +++ b/src/keboola_agent_cli/services/flow_validation.py @@ -165,24 +165,39 @@ def validate_conditional_flow( return _semantic_errors(phases, tasks) -def find_unreachable_phases(phases: list[dict[str, Any]]) -> list[str]: - """Return ids of phases not reachable from the entry phase (first in the - list) by following next[].goto edges. WARNING-level only -- never blocks a - write. Returns ids in the order they appear in ``phases``. +def reachable_phases(phases: list[dict[str, Any]], start_id: str) -> list[str]: + """Return the ids of ``start_id`` and every phase reachable from it. + + Walks ``next[].goto`` edges breadth-first; a ``goto`` of ``None`` is the + flow's END marker and terminates that branch. Ids come back in the order + they appear in ``phases`` (not in visit order), so callers get a stable, + flow-ordered selection regardless of the graph shape. An unknown + ``start_id`` yields an empty list -- callers validate the id themselves so + they can raise with the list of valid ones. """ - if not phases: - return [] by_id = {str(p.get("id")): p for p in phases} - entry = str(phases[0].get("id")) - reachable: set[str] = set() - queue: deque[str] = deque([entry]) + if start_id not in by_id: + return [] + seen: set[str] = set() + queue: deque[str] = deque([start_id]) while queue: pid = queue.popleft() - if pid in reachable or pid not in by_id: + if pid in seen or pid not in by_id: continue - reachable.add(pid) + seen.add(pid) for transition in by_id[pid].get("next", []): goto = transition.get("goto") if goto is not None: queue.append(str(goto)) + return [str(p.get("id")) for p in phases if str(p.get("id")) in seen] + + +def find_unreachable_phases(phases: list[dict[str, Any]]) -> list[str]: + """Return ids of phases not reachable from the entry phase (first in the + list) by following next[].goto edges. WARNING-level only -- never blocks a + write. Returns ids in the order they appear in ``phases``. + """ + if not phases: + return [] + reachable = set(reachable_phases(phases, str(phases[0].get("id")))) return [str(p.get("id")) for p in phases if str(p.get("id")) not in reachable] diff --git a/src/keboola_agent_cli/services/job_service.py b/src/keboola_agent_cli/services/job_service.py index ff57c3db6..34d7505b4 100644 --- a/src/keboola_agent_cli/services/job_service.py +++ b/src/keboola_agent_cli/services/job_service.py @@ -491,6 +491,7 @@ def run_job( mode: str = DEFAULT_JOB_MODE, idempotency_key: str | None = None, force_rerun: bool = False, + only_flow_task_ids: list[str] | None = None, ) -> dict[str, Any]: """Create and optionally wait for a Queue API job. @@ -550,6 +551,11 @@ def run_job( wrong job. force_rerun: If True, ignore any recorded ``idempotency_key`` entry and always create a fresh job (overwriting the stored entry). + only_flow_task_ids: Optional ``keboola.flow`` task-id allowlist for + a partial flow run (issue #725). Resolved by + ``FlowService.resolve_flow_task_ids``; passed through verbatim. + Meaningless for a non-flow component -- the Queue API ignores + it there. Returns: Job dict with ``project_alias``. If wait=True, returns the @@ -605,6 +611,7 @@ def _create() -> dict[str, Any]: branch_id=branch_id, variable_values_id=resolved_values_id, mode=mode, + only_flow_task_ids=only_flow_task_ids, ) store = self._default_idempotency_store() if idempotency_key else None diff --git a/tests/test_flow_run.py b/tests/test_flow_run.py new file mode 100644 index 000000000..a8a9b3601 --- /dev/null +++ b/tests/test_flow_run.py @@ -0,0 +1,361 @@ +"""Tests for partial flow runs -- `kbagent flow run` (issue #725). + +Covers the three layers the feature crosses: the ``onlyFlowTaskIds`` body +field on the Queue client, the phase->task-id resolution in ``FlowService``, +and the ``flow run`` command wiring (including the ``--dry-run`` preview and +the mutually-exclusive selectors). +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from keboola_agent_cli.cli import app +from keboola_agent_cli.config_store import ConfigStore +from keboola_agent_cli.errors import ErrorCode, KeboolaApiError +from keboola_agent_cli.models import ProjectConfig +from keboola_agent_cli.services.flow_service import FlowService +from keboola_agent_cli.services.flow_validation import reachable_phases + +runner = CliRunner() + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + +# A small diamond graph so reachability is actually exercised: +# +# p1 -> p2 -> p4 -> END +# \-> p3 -/ +# +# Every phase carries one task; p3's task is disabled. +_PHASES = [ + {"id": "p1", "name": "Extract", "next": [{"id": "a", "goto": "p2"}, {"id": "b", "goto": "p3"}]}, + {"id": "p2", "name": "Transform", "next": [{"id": "c", "goto": "p4"}]}, + {"id": "p3", "name": "Enrich", "next": [{"id": "d", "goto": "p4"}]}, + {"id": "p4", "name": "Load", "next": [{"id": "e", "goto": None}]}, +] +_TASKS = [ + {"id": "t1", "name": "T1", "phase": "p1", "task": {"type": "job"}}, + {"id": "t2", "name": "T2", "phase": "p2", "task": {"type": "job"}}, + {"id": "t3", "name": "T3", "phase": "p3", "enabled": False, "task": {"type": "job"}}, + {"id": "t4", "name": "T4", "phase": "p4", "task": {"type": "job"}}, +] + + +def _flow_service(phases=None, tasks=None) -> FlowService: + """FlowService whose only live call is get_config_detail.""" + client = MagicMock() + client.get_config_detail.return_value = { + "id": "5", + "name": "CF", + "configuration": { + "phases": _PHASES if phases is None else phases, + "tasks": _TASKS if tasks is None else tasks, + }, + } + cs = MagicMock() + config = MagicMock() + config.projects = { + "prod": MagicMock( + stack_url="https://connection.keboola.com", token="tok", active_branch_id=None + ) + } + config.max_parallel_workers = 10 + cs.load.return_value = config + cs.get_project.side_effect = lambda alias: config.projects.get(alias) + return FlowService( + config_store=cs, + client_factory=lambda url, token: client, + ai_client_factory=lambda url, token: MagicMock(), + scheduler_client_factory=lambda url, token: MagicMock(), + ) + + +# --------------------------------------------------------------------------- +# reachable_phases (pure) +# --------------------------------------------------------------------------- + + +def test_reachable_phases_walks_forward_only(): + assert reachable_phases(_PHASES, "p2") == ["p2", "p4"] + + +def test_reachable_phases_covers_every_branch_of_a_fork(): + assert reachable_phases(_PHASES, "p1") == ["p1", "p2", "p3", "p4"] + + +def test_reachable_phases_returns_flow_order_not_visit_order(): + # BFS from p1 visits p2 before p3, but a graph whose list order differs + # must still come back in list order. + phases = [ + {"id": "z", "name": "Z", "next": [{"id": "n", "goto": None}]}, + {"id": "a", "name": "A", "next": [{"id": "n", "goto": "z"}]}, + ] + assert reachable_phases(phases, "a") == ["z", "a"] + + +def test_reachable_phases_unknown_start_is_empty(): + assert reachable_phases(_PHASES, "nope") == [] + + +def test_reachable_phases_tolerates_a_cycle(): + phases = [ + {"id": "p1", "next": [{"id": "n", "goto": "p2"}]}, + {"id": "p2", "next": [{"id": "n", "goto": "p1"}]}, + ] + assert reachable_phases(phases, "p1") == ["p1", "p2"] + + +# --------------------------------------------------------------------------- +# FlowService.resolve_flow_task_ids +# --------------------------------------------------------------------------- + + +def test_resolve_from_phase_takes_downstream_tasks(): + result = _flow_service().resolve_flow_task_ids(alias="prod", config_id="5", from_phase="p2") + assert result["task_ids"] == ["t2", "t4"] + assert [p["id"] for p in result["selected_phases"]] == ["p2", "p4"] + assert result["from_phase"] == "p2" + + +def test_resolve_from_phase_skips_disabled_tasks_silently(): + result = _flow_service().resolve_flow_task_ids(alias="prod", config_id="5", from_phase="p1") + # t3 is disabled: excluded from the selection, reported separately. + assert result["task_ids"] == ["t1", "t2", "t4"] + assert result["skipped_disabled_task_ids"] == ["t3"] + + +def test_resolve_always_reports_conditions_are_not_evaluated(): + # The single most important field in the payload: a caller must never read + # a selected run as a rehearsal of the flow's condition logic. + result = _flow_service().resolve_flow_task_ids(alias="prod", config_id="5", from_phase="p1") + assert result["conditions_evaluated"] is False + + +def test_resolve_unknown_phase_lists_the_valid_ones(): + with pytest.raises(KeboolaApiError) as exc: + _flow_service().resolve_flow_task_ids(alias="prod", config_id="5", from_phase="nope") + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + assert "p1, p2, p3, p4" in exc.value.message + + +def test_resolve_unreachable_phase_is_rejected_before_the_api_sees_it(): + phases = [ + {"id": "p1", "name": "Entry", "next": [{"id": "n", "goto": None}]}, + {"id": "orphan", "name": "Orphan", "next": []}, + ] + tasks = [{"id": "t9", "name": "T9", "phase": "orphan", "task": {"type": "job"}}] + with pytest.raises(KeboolaApiError) as exc: + _flow_service(phases, tasks).resolve_flow_task_ids( + alias="prod", config_id="5", from_phase="orphan" + ) + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + assert "not reachable" in exc.value.message + + +def test_resolve_phase_whose_downstream_is_all_disabled_errors(): + phases = [{"id": "p1", "name": "P1", "next": [{"id": "n", "goto": None}]}] + tasks = [{"id": "t1", "name": "T1", "phase": "p1", "enabled": False, "task": {"type": "job"}}] + with pytest.raises(KeboolaApiError) as exc: + _flow_service(phases, tasks).resolve_flow_task_ids( + alias="prod", config_id="5", from_phase="p1" + ) + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + assert "no runnable tasks" in exc.value.message + + +def test_resolve_only_task_ids_returns_them_in_flow_order(): + result = _flow_service().resolve_flow_task_ids( + alias="prod", config_id="5", only_task_ids=["t4", "t1"] + ) + # The caller listed t4 first; the run order is the flow's, so t1 comes back + # first rather than echoing the caller's ordering. + assert result["task_ids"] == ["t1", "t4"] + + +def test_resolve_only_task_ids_deduplicates(): + result = _flow_service().resolve_flow_task_ids( + alias="prod", config_id="5", only_task_ids=["t1", "t1"] + ) + assert result["task_ids"] == ["t1"] + + +def test_resolve_only_task_ids_rejects_unknown_id(): + with pytest.raises(KeboolaApiError) as exc: + _flow_service().resolve_flow_task_ids( + alias="prod", config_id="5", only_task_ids=["t1", "ghost"] + ) + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + assert "ghost" in exc.value.message + + +def test_resolve_only_task_ids_rejects_a_disabled_id(): + # Named explicitly => a hard error, unlike the --from-phase silent skip. + with pytest.raises(KeboolaApiError) as exc: + _flow_service().resolve_flow_task_ids(alias="prod", config_id="5", only_task_ids=["t3"]) + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + assert "disabled" in exc.value.message + + +@pytest.mark.parametrize( + "kwargs", + [ + {}, + {"from_phase": "p1", "only_task_ids": ["t1"]}, + ], +) +def test_resolve_requires_exactly_one_selector(kwargs): + with pytest.raises(KeboolaApiError) as exc: + _flow_service().resolve_flow_task_ids(alias="prod", config_id="5", **kwargs) + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + + +# --------------------------------------------------------------------------- +# Queue client: the onlyFlowTaskIds body field +# --------------------------------------------------------------------------- + + +def _created_job_body(**kwargs: Any) -> dict[str, Any]: + """Return the JSON body ``create_job`` would POST for these arguments.""" + from keboola_agent_cli.client import KeboolaClient + + client = KeboolaClient(stack_url="https://connection.keboola.com", token="tok") + with patch.object(client, "_queue_request") as queue_request: + queue_request.return_value.json.return_value = {"id": "123"} + client.create_job(component_id="keboola.flow", config_id="5", **kwargs) + body: dict[str, Any] = queue_request.call_args.kwargs["json"] + return body + + +def test_create_job_sends_only_flow_task_ids(): + body = _created_job_body(only_flow_task_ids=["t2", "t4"]) + assert body["onlyFlowTaskIds"] == ["t2", "t4"] + # The deprecated alias must NOT ride along -- the API rejects both together. + assert "onlyOrchestrationTaskIds" not in body + + +@pytest.mark.parametrize("value", [None, []]) +def test_create_job_omits_the_field_when_unset(value): + assert "onlyFlowTaskIds" not in _created_job_body(only_flow_task_ids=value) + + +# --------------------------------------------------------------------------- +# CLI wiring +# --------------------------------------------------------------------------- + + +def _setup_config(config_dir: Path) -> ConfigStore: + store = ConfigStore(config_dir=config_dir) + store.add_project( + "prod", + ProjectConfig( + stack_url="https://connection.keboola.com", + token="999-token-abc", + project_name="prod", + project_id=1234, + ), + ) + return store + + +def _invoke(store: ConfigStore, flow_service, job_service, args: list[str]): + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlow, + patch("keboola_agent_cli.cli.JobService") as MockJob, + ): + MockStore.return_value = store + MockFlow.return_value = flow_service + MockJob.return_value = job_service + return runner.invoke(app, ["--json", "flow", "run", *args]) + + +def _base_args(*extra: str) -> list[str]: + return ["--project", "prod", "--flow-id", "5", *extra] + + +def test_flow_run_rejects_both_selectors(tmp_path: Path): + store = _setup_config(tmp_path / "cfg") + result = _invoke( + store, + MagicMock(), + MagicMock(), + _base_args("--from-phase", "p2", "--only-task", "t1"), + ) + assert result.exit_code == 2 + assert "mutually exclusive" in result.output + + +def test_flow_run_rejects_dry_run_without_a_selector(tmp_path: Path): + store = _setup_config(tmp_path / "cfg") + result = _invoke(store, MagicMock(), MagicMock(), _base_args("--dry-run")) + assert result.exit_code == 2 + assert "--dry-run needs" in result.output + + +def test_flow_run_dry_run_previews_without_creating_a_job(tmp_path: Path): + store = _setup_config(tmp_path / "cfg") + job_service = MagicMock() + result = _invoke( + store, _flow_service(), job_service, _base_args("--from-phase", "p2", "--dry-run") + ) + assert result.exit_code == 0, result.output + payload = json.loads(result.output)["data"] + assert payload["dry_run"] is True + assert payload["task_ids"] == ["t2", "t4"] + job_service.run_job.assert_not_called() + + +def test_flow_run_passes_resolved_ids_to_the_job_service(tmp_path: Path): + store = _setup_config(tmp_path / "cfg") + job_service = MagicMock() + job_service.run_job.return_value = {"id": "999", "status": "created"} + result = _invoke(store, _flow_service(), job_service, _base_args("--from-phase", "p2")) + assert result.exit_code == 0, result.output + kwargs = job_service.run_job.call_args.kwargs + assert kwargs["component_id"] == "keboola.flow" + assert kwargs["only_flow_task_ids"] == ["t2", "t4"] + # The selection rides along on the result so a caller keeping only the JSON + # can still tell which tasks ran, with their names and phases. + assert json.loads(result.output)["data"]["flow_task_selection"]["task_ids"] == ["t2", "t4"] + + +def test_flow_run_without_a_selector_is_a_plain_full_run(tmp_path: Path): + store = _setup_config(tmp_path / "cfg") + flow_service = MagicMock() + job_service = MagicMock() + job_service.run_job.return_value = {"id": "999", "status": "created"} + result = _invoke(store, flow_service, job_service, _base_args()) + assert result.exit_code == 0, result.output + assert job_service.run_job.call_args.kwargs["only_flow_task_ids"] is None + # No resolution happened at all -- a full run must not fetch the flow body. + flow_service.resolve_flow_task_ids.assert_not_called() + assert "flow_task_selection" not in json.loads(result.output)["data"] + + +def test_flow_run_maps_a_service_error_to_its_exit_code(tmp_path: Path): + store = _setup_config(tmp_path / "cfg") + flow_service = MagicMock() + flow_service.resolve_flow_task_ids.side_effect = KeboolaApiError( + message="Unknown phase id 'nope'.", + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + retryable=False, + ) + result = _invoke(store, flow_service, MagicMock(), _base_args("--from-phase", "nope")) + assert result.exit_code == 2 + assert json.loads(result.output)["error"]["code"] == "INVALID_ARGUMENT" + + +def test_flow_run_is_a_write_operation(): + from keboola_agent_cli.permissions import OPERATION_REGISTRY + + assert OPERATION_REGISTRY["flow.run"] == "write" diff --git a/tests/test_services.py b/tests/test_services.py index 896c8c86d..1c62ec012 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -3475,6 +3475,7 @@ def test_run_job_without_branch(self, tmp_config_dir: Path) -> None: branch_id=None, variable_values_id=None, mode="run", + only_flow_task_ids=None, ) mock_client.close.assert_called_once() @@ -3520,6 +3521,7 @@ def test_run_job_with_branch(self, tmp_config_dir: Path) -> None: branch_id=789, variable_values_id=None, mode="run", + only_flow_task_ids=None, ) def test_run_job_with_branch_and_wait(self, tmp_config_dir: Path) -> None: @@ -3567,6 +3569,7 @@ def test_run_job_with_branch_and_wait(self, tmp_config_dir: Path) -> None: branch_id=123, variable_values_id=None, mode="run", + only_flow_task_ids=None, ) mock_client.wait_for_queue_job.assert_called_once_with( "557", max_wait=60.0, poll_strategy="exponential" @@ -4219,6 +4222,7 @@ def test_run_job_auto_resolves_values_id(self, tmp_config_dir: Path) -> None: branch_id=None, variable_values_id="row-first", mode="run", + only_flow_task_ids=None, ) def test_run_job_explicit_override_wins(self, tmp_config_dir: Path) -> None: