# Sink, notification, secret & cache blocks

Category dirs: `inference/core/workflows/core_steps/sinks/**`, `.../secrets_providers/**`, `.../cache/**`. See the universal invariants section in `SKILL.md` first.

**What / when.** The *terminal / side-effecting* blocks — they push workflow results out to the world (email, Slack, SMS, webhook, local file, Roboflow dataset/monitoring/metadata) or manage runtime state (secrets from env, in-memory cache). Reach for this category when the block's job is a **side effect**, not producing a data kind for downstream steps.

**Mental model.** Every block still subclasses `WorkflowBlock` + `WorkflowBlockManifest` (`inference/core/workflows/prototypes/block.py`) — there is no special "SinkBlock" base. What unifies them is *convention*, not inheritance:
- **`block_type` in `ui_manifest` json_schema_extra** tags the category: `"sink"` (email/webhook/slack/twilio/local_file/dataset_upload/custom_metadata/model_monitoring), `"secrets_provider"` (environment_secrets_store), `"fusion"` (cache_get/cache_set — cache is *not* tagged "sink").
- **Standard sink output contract**: sinks return a fixed dict, almost always `{"error_status": BOOLEAN_KIND, "message": STRING_KIND}` and (for throttled ones) `throttling_status: BOOLEAN_KIND` — see `describe_outputs` in `sinks/webhook/v1.py`. Sinks emit *status*, never data kinds meant to feed compute.
- **The "notification sink" shape** (email/webhook/slack/twilio) is a near-copy-paste family: `disable_sink` early-exit → cooldown check → build a `partial()` handler → dispatch via `BackgroundTasks` / `ThreadPoolExecutor` when `fire_and_forget`, else call sync and return real `error_status`. Compare `run()` in `sinks/webhook/v1.py` and `sinks/email_notification/v2.py` — structurally identical.
- **Message templating + UQL**: notification bodies use `{{ $parameters.x }}` placeholders filled from a `message_parameters` selector dict, optionally transformed per-key via `message_parameters_operations` (UQL op chains, `build_operations_chain`). See `format_email_message`/`execute_operations_on_parameters` in `email_notification/v2.py` and `webhook/v1.py`.
- **Batch/dimensionality**: notification sinks are scalar (one message per run). The *Roboflow* sinks are the batch-oriented outliers — `dataset_upload/v2.py` declares `get_parameters_accepting_batches() -> ["images","predictions","image_name"]` and its `run` takes `Batch[WorkflowImageData]` / `Batch[Union[sv.Detections, dict]]`, iterating with `zip`.
- **Statefulness**: several are stateful and therefore fragile on HTTP — cooldown timestamps (`self._last_notification_fired` in email/webhook), in-memory prediction aggregation (`sinks/roboflow/model_monitoring_inference_aggregator/v1.py`), and the process-global `WorkflowMemoryCache` behind cache_get/set.

**Nuances an author MUST know.**
- **Cooldown is video-only, and you must declare it.** The `cooldown_seconds` timer lives in instance state, so it silently no-ops behind a stateless/multi-replica HTTP server. Notification sinks attach `COOLDOWN_HTTP_SOFT_RESTRICTION` via `get_restrictions()` (`email_notification/v2.py`, `webhook/v1.py`) to surface this. If your sink throttles, declare the restriction.
- **`fire_and_forget=True` destroys your `error_status`.** When dispatched to a background task/thread the block returns `error_status=False` unconditionally — the docstrings explicitly tell users to set `fire_and_forget=False` for debugging (`LONG_DESCRIPTION` in `email_notification/v2.py`). Design outputs knowing async == blind.
- **Secrets: use `SECRET_KIND` and `private=True`.** Password/token fields must accept `Selector(kind=[STRING_KIND, SECRET_KIND])` and set `private=True` so they aren't persisted in the workflow definition — see `sender_email_password` in `email_notification/v2.py` and `slack_token` in `sinks/slack/notification/v1.py`. The `secrets_providers/environment_secrets_store/v1.py` provider is the *source* of `SECRET_KIND`: it declares `describe_outputs -> [OutputDefinition(name="*")]` (wildcard) but resolves concrete lowercased outputs at runtime via `get_actual_outputs()` — the pattern for a **dynamic-output** block.
- **REMOTE execution + statefulness = hard restriction.** Cache and model-monitoring blocks raise `NotImplementedError` outside LOCAL mode; they encode this with a `RuntimeRestriction(severity=Severity.HARD, applies_to_step_execution_modes=[StepExecutionMode.REMOTE])` (`cache/cache_set/v1.py`) and re-check `self._step_execution_mode is not StepExecutionMode.LOCAL` inside `run()`. Env-secrets similarly hard-restricts `HOSTED_SERVERLESS`/`DEDICATED_DEPLOYMENT` runtimes gated on `ALLOW_WORKFLOW_BLOCKS_ACCESSING_ENVIRONMENTAL_VARIABLES`.
- **Cache namespacing + eviction is per-video, via image metadata.** cache_get/set derive the namespace from `image.video_metadata.video_identifier or "default"` and clean up in `__del__` via `WorkflowMemoryCache.clear_namespace(self.namespace)` (`cache_set/v1.py`). Mirror this for any stateful per-stream block: key state by `video_identifier`, evict on destruct. cache values are `WILDCARD_KIND` in/out (pass-through), and a miss returns `False`, not `None`.
- **sv.Detections data density — `inference_id` lives in `.data`.** `sinks/roboflow/custom_metadata/v1.py` pulls `inference_ids = predictions.data.get(INFERENCE_ID_KEY, [])` for `sv.Detections` but `predictions[INFERENCE_ID_KEY]` for a classification dict — two different access paths for the two prediction shapes, and it errors if none are found. A sink needing per-inference identity must know detections carry it in the `.data` dict, not as a top-level field.
- **Image attachments must be serialized by the sink itself.** There's no automatic `WorkflowImageData` → wire conversion. Each sink hand-rolls it: email JPEG-encodes via `encode_image_to_jpeg_bytes` and base64s for the Roboflow proxy, or builds inline-CID MIME parts for SMTP (`serialize_image_data` / `_send_email_using_smtp_server_v2` in `email_notification/v2.py`). Attachment fields accept `Selector(kind=[STRING_KIND, BYTES_KIND, IMAGE_KIND])`.
- **Roboflow-managed vs. custom-credential branching.** email_notification v2 exposes an `email_provider` dropdown; SMTP fields are conditionally shown via `json_schema_extra.relevant_for` and validated at runtime (returns `error_status=True` if required SMTP fields missing rather than raising). This `relevant_for` UI-conditioning pattern is how a sink offers "use Roboflow / bring your own" without separate blocks.

**Example blocks to pattern-match (simplest → most representative).**
- `cache/cache_set/v1.py` (+ `cache/cache_get/v1.py`) — the **stateful, LOCAL-only, video-namespaced** minimal block: `RuntimeRestriction` HARD on REMOTE, `WILDCARD_KIND` pass-through, `__del__` eviction.
- `secrets_providers/environment_secrets_store/v1.py` — a **dynamic-output secrets provider**: `describe_outputs` wildcard + `get_actual_outputs()`, env-gated runtime restriction, `SECRET_KIND`.
- `sinks/webhook/v1.py` — the **canonical scalar notification sink**: disable→cooldown→`partial`→fire-and-forget dispatch, UQL `_operations` on payload dicts, `error_status/throttling_status/message` outputs.
- `sinks/email_notification/v2.py` — **secrets + `relevant_for` provider branching + image/attachment serialization + HTML templating**; the richest notification example.
- `sinks/roboflow/dataset_upload/v2.py` — a **batch-oriented sink**: `get_parameters_accepting_batches`, `Batch[...]` params, `zip` iteration, sampling/quota + `persist_predictions`.
- `sinks/roboflow/model_monitoring_inference_aggregator/v1.py` — **in-memory stateful aggregation over time** (`frequency`, `PredictionsAggregator`, per-run flush) and the paired REMOTE hard-restriction; contrast with `sinks/roboflow/custom_metadata/v1.py` for how `inference_id` is extracted from `sv.Detections.data`.

`sinks/local_file/v1.py` (write CSV/JSON/txt, append-log rotation), `sinks/slack/notification/v1.py`, and `sinks/twilio/sms/v2.py` are the remaining notification-family siblings — pattern-match webhook/email first, then diff.
