Metadata-Version: 2.5
Name: corobots-sdk
Version: 2.1.0
Summary: Python SDK for CoRobots — worker registration, step execution, and workflow management
Project-URL: Homepage, https://github.com/coredgeio/corobots
Project-URL: Repository, https://github.com/coredgeio/corobots
Project-URL: Issues, https://github.com/coredgeio/corobots/issues
Author-email: "Coredge.io" <ashok@coredge.io>
License: Apache-2.0
Keywords: fastapi,grpc,orchestration,saga,workflow
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Framework :: Flask
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: grpcio>=1.60
Requires-Dist: httpx>=0.27
Requires-Dist: protobuf>=6.31.1
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Provides-Extra: dev
Requires-Dist: fastapi>=0.110; extra == 'dev'
Requires-Dist: flask>=2.3; extra == 'dev'
Requires-Dist: grpcio-tools~=1.74.0; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest-httpx>=0.30; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: uvicorn>=0.29; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == 'fastapi'
Provides-Extra: flask
Requires-Dist: flask>=2.3; extra == 'flask'
Description-Content-Type: text/markdown

# corobots-sdk

Python SDK for [CoRobots](https://github.com/coredgeio/corobots):
remote step workers (outbound gRPC stream — no inbound port), workflow definition
building (YAML, `workflow/v1` + `workflow/v2` flow control), workflow/trigger
registration, and execution management.

```bash
pip install corobots-sdk            # core
pip install "corobots-sdk[fastapi]" # + FastAPI integration
pip install "corobots-sdk[flask]"   # + Flask integration
```

```python
import corobots_sdk
```

## Migrating from fluid-workflow-engine-sdk 1.x to corobots-sdk 2.0 <!-- rename-guard -->

2.0 is a clean-break rename of the package itself, with no compatibility
aliases for the old names:

- `pip uninstall fluid-workflow-engine-sdk` (or `-U` over it), then <!-- rename-guard -->
  `pip install corobots-sdk`; `import fluid_workflow_engine_sdk` becomes <!-- rename-guard -->
  `import corobots_sdk`.
- Every `FLUID_*` environment variable becomes `COROBOTS_*` (for example <!-- rename-guard -->
  `FLUID_ENGINE_ADDRESS` → `COROBOTS_ENGINE_ADDRESS`, `FLUID_SERVICE_TOKEN` → <!-- rename-guard -->
  `COROBOTS_SERVICE_TOKEN`) — `COROBOTS_ENGINE_ADDRESS` and
  `COROBOTS_SERVICE_TOKEN` are the same names the engine and the Go SDK use.
  The old `FLUID_*` variables are no longer read at all: they are silently <!-- rename-guard -->
  ignored rather than erroring, so a deployment that only sets the old names
  finds its worker or client silently falling back to `localhost:50052`
  instead of the intended engine address.
- The FastAPI/Flask contrib routes move from `/fluid/*` to `/corobots/*` <!-- rename-guard -->
  (`/fluid/healthz`, `/fluid/steps` → `/corobots/healthz`, `/corobots/steps`). <!-- rename-guard -->
  If you point a Kubernetes liveness or readiness probe at `/fluid/healthz`, <!-- rename-guard -->
  update it to `/corobots/healthz` — the old path now 404s.
- `WorkflowEngineClient` is renamed `CorobotsClient`; update both the import <!-- rename-guard -->
  and the constructor call.
- The deprecated `workflow_engine_sdk` compatibility shim package is removed <!-- rename-guard -->
  entirely — there is no import path left that resolves to it.
- Service tokens now start with `crsvc_`. Old `wfsvc_` tokens are refused by <!-- rename-guard -->
  the engine, so any pre-rename service token must be reissued before a
  worker or client can authenticate.
- The engine's HTTP API prefix is now `/api/v1/corobots`; anything hard-coding
  the old `/api/v1/workflow-engine-admin` prefix must be updated too. <!-- rename-guard -->

## Migrating to 1.0 (worker-initiated step stream)

1.0 is a hard cutover to the engine's `StepStream` dispatch model: the engine
no longer dials workers. A worker registers, then opens one long-lived
bidirectional stream to the engine's registration address over which the
engine pushes step tasks and the worker sends log/result frames back. The
old `StepWorkerService` gRPC server the SDK used to host is gone, and with it
every inbound-connectivity knob:

- `WorkerClient(..., worker_host=..., grpc_port=...)` → removed. The
  constructor is now `WorkerClient(service_name, engine_address, *,
  max_workers=10, tee_logging=True, wait_for_engine_s=0.0, max_inflight=0)`;
  everything after `engine_address` is keyword-only so a stale positional
  `worker_host` fails at construction.
- `WorkerSettings.worker_host` / `grpc_port`, `FLUID_WORKER_HOST` / <!-- rename-guard -->
  `FLUID_GRPC_PORT` and the `SERVICE_HOST` alias → removed; `max_inflight` / <!-- rename-guard -->
  `FLUID_MAX_INFLIGHT` added. Unknown settings fields are now rejected. <!-- rename-guard -->
- `worker.bound_port` → removed; `worker.attached` added (is the step stream
  open right now). `/fluid/healthz` reports `attached` and lists the in-flight <!-- rename-guard -->
  task ids under `active_executions`.
- Steps gain an optional `cancel` kwarg (a `threading.Event` the engine sets
  on cancel / timeout / stream loss) and `delivery_count`.
- No Kubernetes `Service`, no `WORKER_HOST`, no pre-fork port juggling: each
  process (including each Gunicorn / uvicorn child) simply opens its own
  stream after the fork.

Requires an engine that serves `WorkflowRegistrationService.StepStream`.

## Worker quick start

A worker hosts step functions, registers them with the engine, opens the step
stream the engine pushes tasks down, and heartbeats. It never listens for the
engine: the only network requirement is that it can reach `engine_address`.
Steps are plain callables; declare only the keyword args you need — `inputs`,
`workflow_id`, `workflow_name`, `step_name`, `is_compensation`, `retry_count`,
and the optional signature-gated extras `log`, `auth`, `scope_path`, `cancel`
and `delivery_count`.

```python
from corobots_sdk import WorkerClient, RetryableError

worker = WorkerClient(
    service_name="my-svc",
    engine_address="corobots:50052",
)

@worker.step("createThing", description="Creates a thing", default_timeout="30s", max_retries=3)
def create_thing(inputs, log=None, auth=None, cancel=None, **_):
    if log:
        log("INFO", "creating", name=inputs["name"])
    if not_ready():
        raise RetryableError("dependency not ready")   # engine retries per policy
    if cancel is not None and cancel.is_set():          # engine cancelled / timed out
        raise RuntimeError("cancelled")
    return {"id": "thing-123"}                          # step outputs

@worker.step("deleteThing", supports_compensation=True)
def delete_thing(inputs, is_compensation, **_):
    ...
    return {}

worker.start()      # or: app = FastAPI(lifespan=worker.lifespan)
```

Anything logged with the stdlib `logging` module inside a step body is also
streamed to the engine's event log (disable with `WorkerClient(...,
tee_logging=False)`). Caveat: threads spawned *inside* a step body are not
captured — pass `log` explicitly there.

Steps may be `async def` — each invocation runs on a fresh event loop on the
worker's thread pool (`asyncio.run`), so `log`/`auth` and the logging tee work
unchanged. Don't cache loop-bound resources (e.g. a module-level
`httpx.AsyncClient`) across invocations; create them inside the step.

Useful constructor extras: `wait_for_engine_s=30` retries engine registration
with backoff at startup, `max_workers` sizes the thread pool tasks run on, and
`max_inflight` caps how many tasks the engine may push at once (0 = engine
default). Introspection: `worker.worker_id`, `worker.is_running` (registered),
`worker.attached` (step stream open), `worker.active_executions` (task ids in
flight), `worker.step_metadata()`.

### How dispatch works

```
worker ──RegisterWorker──▶ engine :50052      (steps + metadata, then Heartbeat)
worker ──StepStream (bidi)──▶ engine :50052   (hello → welcome, then:)
        ◀── StepTask / StepTaskCancel
        ──▶ StepLogFrame* , StepTaskResult
```

The stream is worker-initiated and long-lived; if it drops (engine restart,
network blip) the engine immediately requeues every task this worker held to
another worker, the SDK cancels the in-flight steps (their `cancel` event is
set and their results are discarded) and reconnects with exponential backoff
(1s → 30s), re-sending its hello. Step functions should therefore be
idempotent or honour `cancel`; a task's `delivery_count` is > 1 when it was
redelivered after a previous worker lost its stream.

Timeouts: when the engine sets a timeout on a task, the SDK reports a
retryable `timed out` failure at the deadline and sets `cancel`; a step that
ignores `cancel` keeps running in the background but its result is discarded.

## Declaring step inputs and outputs

A step can declare the inputs it takes and the outputs it returns. The
schema goes to the engine in `StepMetadata.inputs` / `outputs`, both at
`RegisterWorker` and in every StepStream hello, so every engine replica has it.
Declaring a schema is optional: a step without one behaves as before.

**Explicit field lists.** Pass `inputs=` / `outputs=` a list of `Field`. The
step keeps its plain-dict signature:

```python
from corobots_sdk import Field

@worker.step(
    "resizeVolume",
    inputs=[
        Field("volumeId", type="string", required=True, description="volume to grow"),
        Field("sizeGb", type="int", required=True),
        Field("tier", type="string", default="standard", enum=["standard", "fast"]),
    ],
    outputs=[Field("sizeGb", type="int"), Field("ready", type="bool")],
)
def resize_volume(inputs, **_):
    return {"sizeGb": int(inputs["sizeGb"]), "ready": True}
```

`Field(name, type="any", required=False, default=None, description="", enum=None)`.
`type` is one of `string`, `int`, `number`, `bool`, `map`, `array`, `any` (`""`
means `any`). `default` must be JSON-serialisable (`None` means no default).
`enum` values are compared as strings; booleans become `"true"` / `"false"`. If
you also import `pydantic.Field`, use the `StepField` alias instead. A bad
declaration raises `ValueError` when the decorator runs: an unknown type, an
empty or duplicate name, or a default that isn't JSON-serialisable.

**Pydantic models.** Pass a `BaseModel` subclass as `inputs=` / `outputs=`.
You can also leave them out and annotate the function: if the `inputs`
parameter is annotated with a model, the input schema comes from it. If the
return annotation is a model, so does the output schema.

```python
from typing import Literal, Optional
from pydantic import BaseModel, Field as ModelField

class DeployIn(BaseModel):
    region: str = ModelField(description="target region")
    replicas: int = 1
    size: Literal["small", "large"] = "small"
    note: Optional[str] = None

class DeployOut(BaseModel):
    deployment_id: str
    ready: bool

@worker.step("deploy")                        # schema inferred from the annotations
def deploy(inputs: DeployIn, log, **_) -> DeployOut:
    log("INFO", "deploying", region=inputs.region)
    return DeployOut(deployment_id=f"dep-{inputs.region}", ready=True)
```

When the `inputs` parameter is annotated with a model, the step receives a
validated instance (`Model.model_validate(inputs)`) rather than the dict. If
validation fails, the step fails **without retry** and the error lists each
bad field. A returned model instance is converted with
`model_dump(mode="json", by_alias=True)`, whatever the annotation says. This
works the same for `async def` steps and `StepRouter` steps. An explicit
`inputs=` / `outputs=` value (even `[]`) overrides inference. Inference uses
`typing.get_type_hints`, so annotations written with
`from __future__ import annotations` work as long as the model is defined at
module level. An annotation that can't be resolved means no inference; it
doesn't raise an error.

Model mapping (`schema_from_model(Model)` gives you the list it produces):

| Python annotation | Schema type |
|---|---|
| `str` | `string` |
| `int` | `int` |
| `float` | `number` |
| `bool` | `bool` |
| `dict`, `Mapping[...]`, nested `BaseModel` | `map` |
| `list`, `tuple`, `set`, `Sequence[...]` | `array` |
| `Literal[...]` | the literals' type, plus `enum` of their values |
| `Enum` subclass | `string` (or `int`/`number` for numeric values), plus `enum` |
| `Optional[X]` / `X \| None` | the type of `X` |
| a union of several types, `Any`, anything else | `any` |

`required` comes from `FieldInfo.is_required()`. A field is optional when it
has a default, so the usual `Optional[X] = None` is optional. A static default
(not a `default_factory`) is advertised as the field's `default`. The
description comes from `Field(description=...)`. Field names are the model's
aliases: the validation alias for inputs and the serialization alias for
outputs.

**What the engine enforces.** The rules below apply under the engine's
default `corobots.stepSchema.enforcement: enforce`. With `warn`, the same
violations are only reported as warnings and events. With `off`, nothing is
checked, but declared defaults are still filled in.

- **When a workflow is saved**, its literal step inputs are checked against the
  declared inputs. A missing required input (one without a default) or a
  literal of the wrong type or outside the `enum` rejects the definition.
  Values that are `{{ }}` expressions are only checked for presence. Inputs
  the step doesn't declare get a warning, not an error.
- **Before dispatch**, the engine fills in declared `default`s for inputs that
  are absent and checks the resolved inputs again. A violation fails the step
  without retry, before any worker receives it.
- **Checks are lenient** because workflow inputs often arrive as strings: a
  numeric string passes `int` / `number`, `"true"` / `"false"` (any Go
  `ParseBool` form) passes `bool`, and a JSON object or array string passes
  `map` / `array`.
- **Outputs** are only compared after the step has run. A type mismatch logs a
  warning and never fails the step. Declared outputs aren't required to be
  present.

`worker.step_metadata()` and the `GET <prefix>/steps` routes of the FastAPI
and Flask integrations include each step's `inputs` / `outputs` as lists of
dicts.

## FastAPI integration

```python
from fastapi import FastAPI
from corobots_sdk.contrib.fastapi import worker_lifespan, worker_router

app = FastAPI(lifespan=worker_lifespan(worker))       # starts/stops the worker
app.include_router(worker_router(worker))             # GET /corobots/healthz, /corobots/steps
```

`worker_lifespan` runs the blocking start/stop off the event loop and composes
with an existing lifespan: `worker_lifespan(worker, inner=app_lifespan)`
(worker start → inner enter → serve → inner exit → worker stop).

## Flask integration

```python
from flask import Flask
from corobots_sdk.contrib.flask import FlaskWorker

app = Flask(__name__)
FlaskWorker(worker, app)   # starts the worker now, stops it atexit,
                           # mounts /corobots/healthz + /corobots/steps
```

Under a pre-fork server (Gunicorn): call
`FlaskWorker(worker).init_app(app, start=False)` at import, and start each
fork's worker in a `post_fork` hook —
`app.extensions["corobots_worker"].start()` — so every child opens its own step
stream. Never start in the master under `--preload` (gRPC channels don't
survive `fork()`). See [`examples/gunicorn.conf.py`](examples/gunicorn.conf.py).
The Flask dev-server reloader imports the app twice — run with
`use_reloader=False`.

## Step routers

`StepRouter` decouples step declaration from the worker instance, like
FastAPI's `APIRouter` — feature modules own their steps, `main` assembles:

```python
from corobots_sdk import StepRouter

billing = StepRouter(prefix="billing")   # advertised as "billing.<name>"

@billing.step("charge")
async def charge(inputs, **_):
    return {"chargeId": "..."}

worker.include_router(billing)           # ValueError on duplicate step names
```

Routers nest (`router.include_router(other)`) and can carry workflows
(`router.workflow(defn)`), which register when the including worker starts.

## Workflow auto-registration

```python
worker.workflow(wf)                # WorkflowDefinition, YAML str, or bytes
worker.start()                     # registers steps, then pushes workflows
```

Definitions are pushed right after worker registration
(`source_service=service_name`); a rejected definition raises `EngineError`
and aborts startup. Steps whose `function` is served by this worker get
`executionMode: grpc` defaulted in automatically (the engine dispatches
remotely only on an explicit non-local mode); builtins and other services'
functions are left untouched.

## Configuration via environment

```python
from corobots_sdk import WorkerClient, WorkerSettings

settings = WorkerSettings.from_env()   # COROBOTS_SERVICE_NAME, COROBOTS_ENGINE_ADDRESS,
                                       # COROBOTS_MAX_WORKERS, COROBOTS_TEE_LOGGING,
                                       # COROBOTS_WAIT_FOR_ENGINE_S, COROBOTS_MAX_INFLIGHT
worker = WorkerClient.from_settings(settings)
```

Keyword overrides win over the environment. `COROBOTS_ENGINE_ADDRESS` and
`COROBOTS_SERVICE_TOKEN` are the same variable names the Go SDK and the
builtin worker read.

Set `COROBOTS_SERVICE_TOKEN` (or `service_token=`) to the service token a
Platform Admin created; required once the engine runs with
`auth.serviceAuth: required`. When `WorkerSettings.from_env()` takes the
token from the environment it removes `COROBOTS_SERVICE_TOKEN` from
`os.environ`, so step code and the subprocesses it starts cannot read the
worker's credential (a later `from_env()` in the process still gets it).

TLS to the registration port: `COROBOTS_ENGINE_TLS=true` (or `tls=True`),
`COROBOTS_ENGINE_CA_FILE` / `ca_file=` (unset = system roots),
`COROBOTS_ENGINE_SERVER_NAME` / `server_name=`, and
`COROBOTS_ENGINE_CLIENT_CERT_FILE` / `_KEY_FILE` (`client_cert_file=`,
`client_key_file=`) when the engine asks for a client certificate. Use
`true`/`false` (the Go SDK accepts those but not `yes`/`on`). Both
`WorkerClient` and `CorobotsClient`'s gRPC channel honour them.

## Workflow secrets

A step receives the secrets its definition grants through a `secrets`
keyword argument — a read-only mapping keyed `"ALIAS"` (one key bound) or
`"ALIAS.<key>"` (whole secret bound). Values never travel in `inputs`; see
`docs/WORKFLOW_SECRETS.md` (read its **security model** first).

```python
@worker.step("terraform.apply")
def apply(inputs, secrets, log, **_):
    token = secrets["TF_TOKEN"]            # KeyError("TF_TOKEN") when not granted
    log("INFO", "applying")                # values of 4+ bytes are masked as *** in log
    return {"applied": True}               # frames, outputs and error text (engine masks again)
```

The engine delivers secret-bearing tasks only to workers attached **over
the engine's own TLS** with a valid **service token** and the `secrets.v1`
capability (announced by this SDK). Without them the task waits
(`step.queued` `waitingFor: secure-worker`) and expires after the queue TTL.
Transformed values (base64, substrings) are not masked — never log
secrets. Definitions grant them with `Step.secret(alias, ref, key=None)` /
`Compensation.secret(...)`:

```python
from corobots_sdk.definition import Compensation, Step

Step("deploy", "terraform.apply").execution_mode("grpc") \
    .secret("AWS", "project:aws-creds") \
    .secret("TF_TOKEN", "platform:tf-cloud", key="token") \
    .compensation(Compensation("terraform.destroy", execution_mode="grpc")
                  .secret("AWS", "project:aws-creds"))
```

Registering a definition that declares secrets over gRPC needs the service
token even when the engine runs `auth.serviceAuth: optional`.

## Building workflow definitions

```python
from corobots_sdk import WorkflowDefinition, Step, RetryPolicy

wf = (
    WorkflowDefinition("provision-fleet")
    .api_version("workflow/v2")            # required for flow-control operators
    .input("regions", type="list", required=True)
    .input("env", type="string", default="staging")
    .step(
        Step("deployAll")
        .foreach("inputs.regions", as_="region", parallel=True, max_concurrency=4)
        .body(
            Step("provision", function="createVpc")
            .input("region", "{{ loop.region }}")
            .retry(RetryPolicy(max_retries=3, initial_backoff="1s"))
            .compensation("deleteVpc", inputs={"region": "{{ loop.region }}"}),
        )
    )
    .step(
        Step("notify")
        .if_("inputs.env == 'prod'")
        .body(Step("page", function="pageOncall"))
        .else_(Step("slack", function="notifySlack"))
        .depends_on("deployAll")
    )
    .output("done", "{{ steps.deployAll.outputs.completed }}")
)
print(wf.to_yaml())
```

`while_`/`until` (polling loops, `max_iterations` mandatory) and
`switch`/`case`/`default` are also available. Step inputs support the object
form for optional values: `Step(...).input("size", "{{ inputs.size }}",
required=False, default="m5.large")`.

## Schedules

A schedule starts a workflow on a timetable: the union of cron
expressions, fixed intervals and Temporal-style calendars, minus
exclusions, in a time zone, optionally bounded and jittered. Three kinds
exist, told apart by `owner_type`:

| Owner | Declared by | Edit / delete via |
|---|---|---|
| `user` | `CorobotsClient.create_schedule` (REST, as your API-token user) | REST |
| `system` | a service over gRPC: `WorkerClient.schedule` / `@worker.scheduled`, `CorobotsClient.register_schedule` | the declaring service |
| `workflow` | the workflow's own YAML (`spec.schedules`, `Schedule(...)` in the builder) | the workflow definition |

Any schedule can be paused, unpaused and triggered by whoever may operate
it; editing or deleting one you do not own fails with `EngineError`
(`status_code` 409, `code` `"schedule_owned"`, `owner_type`). A schedule's
runs start as a dedicated schedule principal, not as the user who
created it. The engine-side reference (spec semantics, DST, policies,
exactly-once, HTTP/gRPC tables) is
[`docs/WORKFLOW_SCHEDULES.md`](../docs/WORKFLOW_SCHEDULES.md). Durations are Go duration strings (`"90s"`, `"1h30m"`) or
`datetime.timedelta`; times are RFC3339 (a naive `datetime` is UTC).

### The spec

```python
from datetime import timedelta
from corobots_sdk import CalendarSpec, IntervalSpec, ScheduleSpec

spec = ScheduleSpec(
    cron=["0 2 * * MON-FRI", "CRON_TZ=UTC @hourly"],   # 5/6 fields, macros, CRON_TZ=
    intervals=[IntervalSpec(every=timedelta(hours=1), offset="15m")],
    calendars=[CalendarSpec(hour="9,17", minute=30, day_of_week="MON-FRI")],
    exclude=[CalendarSpec(month=12, day_of_month=25)],
    timezone="Asia/Kolkata",                            # IANA, default UTC
    jitter="30s",                                       # deterministic per fire
)
```

`SchedulePolicy(overlap=..., catchup_window=..., pause_on_failure=...)`
controls what happens when a fire meets a running run — `OverlapPolicy`
`skip` (default), `bufferOne`, `bufferAll`, `cancelOther` (no
compensation) or `allowAll` — and how late a missed fire may still start
(`"0s"` = never; unset = the engine default).

### From a worker (system-owned)

```python
worker = WorkerClient("billing-svc", "engine:50052", service_token="crsvc_...")

@worker.scheduled("nightly-invoices", workflow="invoice-run",
                  cron="0 2 * * *", timezone="Europe/London",
                  inputs={"day": "{{ schedule.scheduledTime }}"})
@worker.step("billInvoices")
def bill(inputs, **_): ...

worker.schedule(ScheduleDefinition(name="hourly-sync", workflow_name="sync",
                                   spec=ScheduleSpec(intervals=[{"every": "1h"}])))
worker.start()   # steps, then workflows, then schedules
print(worker.schedule_results)   # {"nightly-invoices": RegisterScheduleResult(unchanged=...), ...}
```

Declarations are validated when made and pushed with `RegisterSchedule`
after the workflows, keyed by (service name, `project_id`, name). An
unchanged declaration is a no-op (`unchanged=True`), so every replica can
declare the same schedules on every start; `replace=True` (the default)
updates one whose definition changed. The engine only accepts schedule
registrations that carry a `crsvc_` service token, even under
`auth.serviceAuth: optional`. As in the Go SDK, a failed registration is
logged and recorded in `schedule_results` without aborting startup. A
runtime pause survives re-registration (`paused=` is the initial state
only). `StepRouter` has the same `schedule` / `scheduled` pair, merged by
`include_router`.

### Over REST (user-owned)

```python
from corobots_sdk import CorobotsClient, ScheduleDefinition, ScheduleSpec

with CorobotsClient("http://engine:50051", api_token="crp_...", project_id="proj-123") as c:
    print(c.preview_schedule(ScheduleSpec(cron=["0 9 * * MON-FRI"]), count=5).summary)
    s = c.create_schedule(ScheduleDefinition(
        name="weekday-report", workflow_name="report",
        spec=ScheduleSpec(cron=["0 9 * * MON-FRI"], timezone="Asia/Kolkata"),
        inputs={"format": "pdf"},
    ))
    d = s.to_definition(); d.remaining_actions = 20
    c.update_schedule(s.id, d)
    c.pause_schedule(s.id, note="holiday"); c.unpause_schedule(s.id)
    run = c.trigger_schedule(s.id, idempotency_key="report-2026-10-01")  # retry-safe
    runs = c.list_executions(schedule=s.id)
    c.delete_schedule(s.id)
```

Also `get_schedule` (next fires, recent actions, running runs, principal)
and `list_schedules(workflow=, status=, owner=)`. Creating, editing,
pausing and triggering need the project admin role. Definitions are
checked locally first (`InvalidScheduleError`, a `ValueError`, listing
field paths); the engine validates the rest and answers 400 with
`EngineError.fields`.

### Over gRPC (service token)

`register_schedule(defn, source_service=, replace=)` and
`unregister_schedule(name, source_service=, project_id=)` manage a
service's own schedules; `list_schedules_grpc`, `describe_schedule_grpc`,
`pause_schedule_grpc`, `unpause_schedule_grpc`, `trigger_schedule_grpc`
and `preview_schedule_grpc` act on any schedule by id, like the Go SDK.
Mutations need the client's `service_token=`; a refused call raises
`EngineError` with `grpc_code`, a rejected registration with the
engine's message.

### In the workflow definition (workflow-owned)

```python
from corobots_sdk import Schedule

wf = wf.schedule(
    Schedule("nightly")
    .cron("0 2 * * *").timezone("Asia/Kolkata").jitter("2m")
    .overlap("skip").catchup_window("30m")
    .input("regions", ["ap-south-1"])
)
```

The engine reconciles `spec.schedules` into workflow-owned schedules in
the project the workflow is registered to, whenever the definition's
content changes. `wf.content_hash()` is the engine's canonical hash of
the definition — the `content_hash` `register_workflow` returns — and is
the same for the same workflow built with the Go builder or written by
hand. (`WorkerClient.workflow()` adds `executionMode: grpc` to the
worker's own steps, which is a different definition.) `to_yaml()` quotes
strings the engine's YAML parser would read as another type (`y`, `1e5`,
`0o17`), so schedule inputs keep their Python types.

## Messages and signals

A `waitFor` step parks a run until a **message** (addressed by name and
correlation key, delivered to one waiting run, buffered for its `ttl`
when none waits yet) or a **signal** (by name, resumes every run waiting
at that moment, never buffered) arrives. Publish them with the client:

```python
from datetime import timedelta
from corobots_sdk import CorobotsClient

with CorobotsClient("http://engine:50051", api_token="crp_...", project_id="proj-123") as client:
    res = client.publish_message(
        "approval.decided", "REQ-1042", {"approved": True},
        message_id="approval-REQ-1042",   # idempotent: a retry returns outcome "duplicate"
        ttl=timedelta(hours=24),          # or "24h", or seconds; 0 = never buffer; None = engine default
    )
    print(res.outcome)                    # delivered | buffered | started | dropped | duplicate

    sig = client.broadcast_signal("maintenance.over", {"window": "w1"}, signal_id="mw-1")
    print(sig.resumed, sig.draining)
```

`publish_message` / `broadcast_signal` go over REST (`POST /messages`,
`POST /signals`) and need any project role (a Platform Admin for platform
scope); `project_id=` overrides the client's project for one call, and
`""` selects platform scope. `publish_message_grpc` /
`broadcast_signal_grpc` take the same arguments and go over gRPC with the
client's `service_token=`, which they require whatever
`auth.serviceAuth` is. They return `MessageResult(message_id, outcome,
run_id)` and `SignalResult(signal_id, resumed, draining, duplicate)`.
Bad input raises `InvalidMessageError` (a `ValueError`) before anything
is sent; engine refusals raise `EngineError` (`status_code` 400 / 403 /
409 / 503, or `grpc_code`). Reusing an id with a different payload is a
409 / `ALREADY_EXISTS`. See [`examples/publish_message.py`](examples/publish_message.py).

## Engine client

```python
from corobots_sdk import CorobotsClient, TriggerSpec

with CorobotsClient("http://engine:50051", engine_grpc="engine:50052") as client:
    client.register_workflow(wf, replace=True, source_service="my-svc")
    res = client.start_workflow(
        "provision-fleet",
        {"regions": ["us-east-1"]},
        started_by="user:ashok",
        tenant="acme", domain="default", project="demo",
    )
    detail = client.get_execution(res.workflow_id)

    client.register_trigger(
        TriggerSpec(
            name="on-vm-delete",
            resource_type="compute",
            event_type="deleted",
            workflow_name="cleanup-vm",
            input_mappings={"vmName": "event.resource_name"},
        ),
        source_service="my-svc",
        replace=True,
    )
```

Schedules have their own methods — see "Schedules" above.

Also available: `unregister_workflow`, `list_workflow_definitions` (gRPC
registry view), `resume_from_step(workflow_id, step_name)`,
`unregister_trigger`, `list_triggers`, `list_audit_events(**filters)` (HTTP
`GET /audit-events`, newest first — see below), plus the HTTP admin surface
(`list_workflows`, `get_workflow`, `list_executions`, `get_execution`,
`cancel_execution`).

### Authenticating the HTTP admin surface

A personal API token — minted in the web UI under **Account → API
tokens** — authenticates you as yourself against the HTTP admin surface
(`list_workflows`, `get_execution`, `list_audit_events`, ...). Pass it as
`api_token=`, and pair it with `project_id=` to scope every call to one
project:

```python
client = CorobotsClient(
    "http://engine:50051",
    api_token="crp_...",
    project_id="proj-123",
)
```

`api_token` is sent as `Authorization: Bearer <token>` and `project_id` as
`X-Project-Id` on every HTTP request. Changing your account password
revokes all of your personal API tokens, so a running client will need a
freshly minted one afterwards.

Workers authenticate differently — with a `service_token=` (or
`COROBOTS_SERVICE_TOKEN`, see "Configuration via environment"
above), presented as gRPC metadata rather than an HTTP header.

The engine no longer honours a `Ce-Userinfo` identity header — there is no
gateway mode. `headers=` still exists for anything else you need sent on
every request (it always takes precedence over anything `api_token=`/
`project_id=` would otherwise add), but authenticating the HTTP admin
surface always means a session cookie or an `api_token=`.

### Audit events

```python
page = client.list_audit_events(project_id="proj-123", action="workflow.register", limit=100)
for event in page.events:
    print(event.ts, event.actor.username, event.action, event.target.name, event.outcome)

if page.next_before:  # more rows follow; page backward in time
    older = client.list_audit_events(before=page.next_before, before_id=page.next_before_id)
```

Access is scoped server-side to what the caller (`api_token=`) is allowed to
see: Platform Admins see every project (or one, via `project_id=`); everyone
else only the projects where they hold the Admin role.

## Examples

See [`examples/`](examples/) for runnable scripts: basic worker, FastAPI
(plain + router/settings/composed-lifespan), Flask (+ Gunicorn config),
workflow auto-registration, definition building with flow control and
schedules, register-and-start, trigger registration, schedule management
over REST, worker-declared schedules and publishing an approval message. Framework guides:
[`docs/PYTHON_FASTAPI_WORKER.md`](docs/PYTHON_FASTAPI_WORKER.md) and
[`docs/PYTHON_FLASK_WORKER.md`](docs/PYTHON_FLASK_WORKER.md).

## Development

```bash
pip install -e ".[dev]"
make gen-stubs         # regenerate gRPC stubs from ../../api/workflow/workflow_service.proto
make test              # unit + offline integration tests (FakeEngine, no Docker)
make test-integration  # just the offline FakeEngine integration tier
make e2e-up            # MongoDB via docker compose (needs `make build` at repo root)
make test-e2e          # real engine + Mongo end-to-end tier
make e2e-down
```

`grpcio-tools` is pinned so regenerated stubs keep the same protobuf gencode
version as the committed ones (protobuf 6.31.x); if you bump it, bump the
`protobuf` runtime floor in `pyproject.toml` to match the new gencode
requirement.

### Releasing

1. Bump `version` in `pyproject.toml` and add a `## [X.Y.Z]` entry to `CHANGELOG.md`.
2. Tag and release: `git tag sdks/python/vX.Y.Z && git push origin sdks/python/vX.Y.Z`, then create a
   GitHub release from that tag. `.github/workflows/sdk-python-publish.yml` builds and uploads to PyPI.
   Tags of the old form `python-sdk/vX.Y.Z` no longer trigger a publish.
