feat(job): client-side idempotency key for run_job (#427)

CLAUDE.md

@@ -309,7 +309,7 @@ kbagent search QUERY [--project NAME] [--type table|bucket|config|flow|data-app|
 
 kbagent job list [--project NAME] [--component-id ID] [--status STATUS] [--limit N]
 kbagent job detail --project NAME --job-id ID
-kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N]
+kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] [--idempotency-key KEY] [--force-rerun]
 kbagent job terminate --project NAME (--job-id ID [--job-id ID ...] | --status any|created|waiting|processing [--component-id ID] [--config-id ID] [--branch ID] [--limit N]) [--dry-run] [--yes]
 
 kbagent storage buckets [--project NAME] [--branch ID]

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

@@ -94,7 +94,7 @@ Requires a **super-admin** Manage API token (same kind as `org setup`). Same def
 ## Job History
 - `job list [--project NAME] [--component-id ID] [--config-id ID] [--status STATUS] [--limit N]` -- list jobs (default 50, max 500)
 - `job detail --project NAME --job-id ID` -- full job detail with timing and result message
-- `job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N]` -- run a job, optionally wait for completion (branch-aware). For configs with linked `keboola.variables` (root-level `configuration.variables_id`), kbagent auto-resolves a `variableValuesId` so transformations bind to the deployed values row. `--variable-values-id` overrides; `--no-variables` skips resolution. `NO_VARIABLE_ROWS` when the linked variables config has zero rows -- fix via `kbagent config variables-set`. `--mode debug` (since v0.43.6) sets the Queue API job `mode` body field to `"debug"`: the component runs with the same configuration + inputs but the worker redirects output to a Storage File tagged `debug-<jobId>` instead of writing to destination buckets. Use for dry-runs, reproducing a failing job on a production config without touching downstream tables, or harvesting the worker's output bytes (download via `storage file-download --tag debug-<jobId>`) to feed into VCR fixtures or component test cases. Default `--mode run` is unchanged. Invalid values (`--mode anything-else`) exit 2 at the Click choice gate before any wire call. Under `--wait`, polls with an exponential curve (2s x 30 -> 5s x 48 -> 15s); `--poll-strategy fixed` keeps a constant 1s interval. On FAILED/WARNING/TERMINATED, the last `--log-tail-lines` events (default 200, **0 disables -- recommended for automation pipelines**) are attached as `logTail` in the JSON result (or `details.logTail` on errors). If `--timeout` expires, kbagent issues `kill_job` on the remote and exits **7** (`JOB_TIMEOUT_TERMINATED`) with the cancelled `details.job` + `details.logTail`; if the kill itself fails, exits **4** (`QUEUE_JOB_TIMEOUT`, `retryable=true`). Use jq pattern `.error.details.logTail? // .data.logTail? // []` to pick up the tail regardless of exit code.
+- `job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] [--idempotency-key KEY] [--force-rerun]` -- run a job, optionally wait for completion (branch-aware). `--idempotency-key KEY` (since v0.63.0) makes a replayed run safe: on a second `job run` with the same key, a prior still-running or non-failed job is returned (JSON gains `idempotent_replay: true`, human mode prints a note) instead of creating a duplicate side effect; a prior FAILED run is re-run. Dedup is **client-side** (the Queue API has no server idempotency token -- verified against the live spec) and persisted to `<config-dir>/job_idempotency.json`, so it is scoped to one machine. Reusing a key for a *different* component/config exits with `INVALID_ARGUMENT` rather than returning the wrong job; `--force-rerun` ignores the stored entry and always creates a fresh job. For configs with linked `keboola.variables` (root-level `configuration.variables_id`), kbagent auto-resolves a `variableValuesId` so transformations bind to the deployed values row. `--variable-values-id` overrides; `--no-variables` skips resolution. `NO_VARIABLE_ROWS` when the linked variables config has zero rows -- fix via `kbagent config variables-set`. `--mode debug` (since v0.43.6) sets the Queue API job `mode` body field to `"debug"`: the component runs with the same configuration + inputs but the worker redirects output to a Storage File tagged `debug-<jobId>` instead of writing to destination buckets. Use for dry-runs, reproducing a failing job on a production config without touching downstream tables, or harvesting the worker's output bytes (download via `storage file-download --tag debug-<jobId>`) to feed into VCR fixtures or component test cases. Default `--mode run` is unchanged. Invalid values (`--mode anything-else`) exit 2 at the Click choice gate before any wire call. Under `--wait`, polls with an exponential curve (2s x 30 -> 5s x 48 -> 15s); `--poll-strategy fixed` keeps a constant 1s interval. On FAILED/WARNING/TERMINATED, the last `--log-tail-lines` events (default 200, **0 disables -- recommended for automation pipelines**) are attached as `logTail` in the JSON result (or `details.logTail` on errors). If `--timeout` expires, kbagent issues `kill_job` on the remote and exits **7** (`JOB_TIMEOUT_TERMINATED`) with the cancelled `details.job` + `details.logTail`; if the kill itself fails, exits **4** (`QUEUE_JOB_TIMEOUT`, `retryable=true`). Use jq pattern `.error.details.logTail? // .data.logTail? // []` to pick up the tail regardless of exit code.
 - `job terminate --project NAME (--job-id ID [--job-id ...] | --status any|created|waiting|processing [--component-id ID] [--config-id ID] [--branch ID] [--limit N]) [--dry-run] [--yes]` -- kill running Queue API jobs. Use to stop runaway loops or clean up pile-ups from repeated `job run` calls. Two modes: by ID (single/batch) or by filter (`--status any` catches every killable state). Response partitions IDs into `killed / already_finished / not_found / failed`; safe to re-run idempotently. Kill is async -- poll `job detail` for `isFinished=true`.
 
 ## Storage

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

@@ -2700,3 +2700,35 @@ hits the SYNCHRONOUS Storage endpoint (no job to poll). `--not-null` on a table
 that ALREADY HAS ROWS is rejected by the backend with an API error (not a local
 validation error) unless you also pass `--default` -- the existing rows need a
 value for the new non-null column. Add `--default` when the table is non-empty.
+
+### `job run --idempotency-key` is client-side dedup, scoped to one machine (since v0.63.0)
+
+The Keboola Queue API `POST /jobs` accepts **no** client-supplied idempotency /
+dedup token -- verified against the live OpenAPI spec (v1.3.8) and the server
+source (an internal `deduplicationId` exists but is daemon-only, never read from
+the public create-job request). So `kbagent job run --idempotency-key KEY`
+de-duplicates **client-side**: a `<config-dir>/job_idempotency.json` map of
+`key -> prior job`. Consequences to know:
+- Dedup only works where that file is shared -- it is per config-dir / per
+  machine. A replay from a *different* machine is NOT deduplicated.
+- A prior run that is still running or finished non-failed is returned
+  (`idempotent_replay: true`); a prior FAILED run is re-run; `--force-rerun`
+  always creates fresh.
+- Reusing a key for a *different* `--component-id`/`--config-id` exits
+  `INVALID_ARGUMENT` (it refuses to return the wrong job), not silently.
+- In-process SDK users get the same via `Client.run_job(idempotency_key=...,
+  idempotency_store=JobIdempotencyStore(path))` -- the facade is config-dir-free,
+  so you must supply the store path.
+
+### The importable SDK is now typed (`py.typed` + result models) (since v0.63.0)
+
+`keboola_agent_cli` ships a PEP 561 `py.typed` marker, so `mypy`/`ty`/IDEs treat
+the in-process library as typed. The high-traffic facade methods return pydantic
+models (`JobResult`, `QueryResult`, `UploadTableResult`, `ConfigDetailResult`,
+`SyncPushResult`) exported from the package root, instead of bare dicts. Every
+model is `extra="allow"`, so a new backend field never raises -- the *named*
+fields are the stable, semver-versioned surface and extras stay reachable via
+attribute access / `model_dump()`. They also accept the raw API key or the
+snake_case field name, so `JobResult.model_validate(service_dict)` works on a
+service-layer dict directly. This is a typing/contract addition only; the dict
+shapes returned by the service layer and the `--json` CLI output are unchanged.

plugins/kbagent/skills/kbagent/references/library-workflow.md

@@ -13,10 +13,11 @@ shell operations, use the `kbagent` CLI.
 
 | Symbol | Purpose |
 |--------|---------|
-| `Client(url, token, *, branch_id=None)` | Stateless entry point to one project; context manager |
+| `Client(url, token, *, branch_id=None, idempotency_store=None)` | Stateless entry point to one project; context manager |
 | `Client.query(workspace_id, sql, *, transactional=False, limit=500)` | Run SQL in a workspace -> `list[dict]` |
 | `Client.query_result(workspace_id, sql, ...)` | Same, but typed -> `QueryResult` (columns + truncation) |
-| `Client.run_job(component_id, config_id, *, wait=False, ...)` | Run a Queue job -> `JobResult` |
+| `Client.run_job(component_id, config_id, *, wait=False, idempotency_key=None, ...)` | Run a Queue job -> `JobResult` (replay-safe with a key) |
+| `JobIdempotencyStore(path)` | Client-side dedup map for replay-safe `run_job` |
 | `Client.config_detail(component_id, config_id, *, branch_id=None)` | One config's detail -> `ConfigDetailResult` |
 | `Client.upload_table(table_id, file_path, *, incremental=False, ...)` | Import a CSV into an existing table -> `UploadTableResult` |
 | `Client.files.upload(source, *, name=None, tags=None, permanent=False)` | Upload a path **or** bytes -> `FileEntry` |
@@ -26,12 +27,12 @@ shell operations, use the `kbagent` CLI.
 | `Client.raw` | The underlying `KeboolaClient` for endpoints the facade omits |
 | `FileEntry` | Uniform file shape: `id, name, tags, created, size_bytes, is_permanent, raw` |
 
-Everything exported from `keboola_agent_cli` (`Client`, `Files`, `FileEntry`, and
-the typed result models `JobResult`, `QueryResult`, `UploadTableResult`,
-`SyncPushResult`, `ConfigDetailResult`) is committed public API and follows
-semver. Since 0.63.0 the package ships a **`py.typed`** marker (PEP 561), so
-`mypy` / `ty` / IDEs treat the SDK as typed -- a contract change surfaces at
-type-check time, not at runtime.
+Everything exported from `keboola_agent_cli` (`Client`, `Files`, `FileEntry`,
+`JobIdempotencyStore`, and the typed result models `JobResult`, `QueryResult`,
+`UploadTableResult`, `SyncPushResult`, `ConfigDetailResult`) is committed public
+API and follows semver. Since 0.63.0 the package ships a **`py.typed`** marker
+(PEP 561), so `mypy` / `ty` / IDEs treat the SDK as typed -- a contract change
+surfaces at type-check time, not at runtime.
 
 ## Typed result models
 
@@ -56,6 +57,38 @@ are the stable surface, but anything else the API returns is preserved
 never raises. They also accept the raw API key *or* the snake_case field name, so
 `JobResult.model_validate(service_dict)` works directly on a service-layer dict.
 
+## Replay-safe job runs (idempotency)
+
+An agentic orchestrator that replays a side-effecting build step after a crash
+must not fire the same job twice. The Queue API has **no** server-side
+idempotency token, so kbagent dedups client-side: give `run_job` an
+`idempotency_key` and a `JobIdempotencyStore` (the facade is config-dir-free, so
+*you* choose where the dedup map lives -- typically inside your resume-checkpoint
+dir).
+
+```python
+from keboola_agent_cli import Client, JobIdempotencyStore
+
+store = JobIdempotencyStore("/var/run/myapp/job_idempotency.json")
+with Client(url=URL, token=TOKEN, idempotency_store=store) as kbc:
+    job = kbc.run_job("keboola.ex-db-snowflake", "12345",
+                      idempotency_key="bootstrap-extract", wait=True)
+    if job.idempotent_replay:
+        ...  # a prior run was returned -- no new job fired
+```
+
+- A prior run that is still running or finished **non-failed** is returned
+  (`job.idempotent_replay is True`); a prior **failed** run is re-run;
+  `force_rerun=True` always creates fresh.
+- Reusing a key for a *different* component/config raises (it refuses to return
+  the wrong job).
+- Dedup is scoped to the store file -- a replay from a machine that does not
+  share it is **not** deduplicated.
+- Pass `idempotency_store=` per-call to override the constructor's; passing
+  `idempotency_key` with no store anywhere raises `ValueError` (the stateless
+  facade has no config-dir to default it to). The `kbagent job run
+  --idempotency-key` CLI path defaults the store to `<config-dir>/`.
+
 ## Auth & construction
 
 Auth is the storage token you pass in (12-factor) -- nothing is persisted to disk.

src/keboola_agent_cli/__init__.py

@@ -11,6 +11,7 @@
     SyncPushResult,
     UploadTableResult,
 )
+from .services.job_idempotency_store import JobIdempotencyStore
 
 try:
     __version__ = version(APP_NAME)
@@ -22,6 +23,7 @@
     "ConfigDetailResult",
     "FileEntry",
     "Files",
+    "JobIdempotencyStore",
     "JobResult",
     "QueryResult",
     "SyncPushResult",

src/keboola_agent_cli/commands/context.py

@@ -266,7 +266,7 @@
   kbagent job detail --project NAME --job-id ID
     Full job detail including result message and timing.
 
-  kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N]
+  kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] [--idempotency-key KEY] [--force-rerun]
     Run a Queue API job. --row-id selects specific config rows (repeatable; omit to run entire config).
     --wait polls until job finishes. --timeout sets max wait in seconds (default 300). Branch-aware.
     When the config has linked variables (configuration.variables_id), kbagent auto-resolves

src/keboola_agent_cli/commands/job.py

@@ -238,6 +238,23 @@ def job_run(
             f"{MAX_LOG_TAIL_LINES}."
         ),
     ),
+    idempotency_key: str | None = typer.Option(
+        None,
+        "--idempotency-key",
+        help=(
+            "Client-supplied de-duplication token (issue #427). On replay with "
+            "the same key, a prior still-running or non-failed job is returned "
+            "instead of creating a duplicate -- safe for resumed/retried build "
+            "steps. A prior FAILED run is re-run. Dedup is client-side (the "
+            "Queue API has no idempotency key) and scoped to this machine's "
+            "config-dir; reusing a key for a different component/config errors."
+        ),
+    ),
+    force_rerun: bool = typer.Option(
+        False,
+        "--force-rerun",
+        help="Ignore any stored --idempotency-key entry and always create a fresh job.",
+    ),
 ) -> None:
     """Run a job for a component configuration.
 
@@ -328,6 +345,8 @@ def job_run(
             poll_strategy=poll_strategy,
             log_tail_lines=log_tail_lines,
             mode=mode,
+            idempotency_key=idempotency_key,
+            force_rerun=force_rerun,
         )
     except ConfigError as exc:
         formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR)
@@ -347,6 +366,11 @@ def job_run(
     if formatter.json_mode:
         formatter.output(result)
     else:
+        if result.get("idempotent_replay"):
+            formatter.console.print(
+                "[dim]Idempotency key matched a prior run -- returning the existing "
+                "job (no new job created).[/dim]"
+            )
         resolved_id = result.get("resolvedVariableValuesId")
         if resolved_id:
             formatter.console.print(f"[dim]Bound variable values row: {escape(resolved_id)}[/dim]")

src/keboola_agent_cli/constants.py

@@ -115,6 +115,10 @@
 # failures without touching production tables.
 VALID_JOB_MODES: frozenset[str] = frozenset({"run", "debug"})
 DEFAULT_JOB_MODE: str = "run"
+# Client-side job idempotency store (issue #427). Lives in the config-dir
+# alongside config.json; maps idempotency_key -> prior job so a replayed
+# `kbagent job run --idempotency-key` does not fire a duplicate side effect.
+JOB_IDEMPOTENCY_FILENAME: str = "job_idempotency.json"
 # Default log-tail length surfaced on FAILED/WARNING/TERMINATED jobs.
 DEFAULT_LOG_TAIL_LINES: int = 200
 # Upper bound to prevent accidentally pulling tens of thousands of events