Metadata-Version: 2.4
Name: allus-company-data
Version: 0.1.2
Summary: Reference Python SDK for the allus company-data API: typed, plaintext, slug-keyed conclusions with transparent decryption.
Author: allme.fyi
License: MIT
Project-URL: Homepage, https://github.com/allus-fyi/company-data-python
Project-URL: Repository, https://github.com/allus-fyi/company-data-python
Keywords: allus,allme,company-data,sdk
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: cryptography>=42
Requires-Dist: requests>=2.31
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Dynamic: license-file

# allus-company-data (Python)

The Python SDK for the **allus company-data API**. Point it at a JSON
config file and it hands back typed, plaintext, **your-slug-keyed conclusions**:
for each connected person, a map of *your request-field slug → plaintext value*
(plus whether the value is live and when it last changed).

The SDK hides everything else — the OAuth token, the field catalog, the id
plumbing, the hybrid decryption, binary fetching, the changes-queue mechanics,
JSON-vs-XML. The platform is **zero-knowledge**: the API only ever holds
ciphertext, so all decryption happens inside the SDK with your service private
key. **The person's own field choices are never exposed** — you only ever see
the request slots you configured.

> This SDK is one of six language ports that share an identical API surface.
> This manual is the Python view of it.

**Contents:** [TL;DR — fetch new updates](#tldr--fetch-new-updates) ·
[Quickstart](#quickstart) · [Every call](#every-call) ·
[The typed value model](#the-typed-value-model) ·
[The changes pump](#the-changes-pump) · [Webhooks](#webhooks) ·
[Company documents](#company-documents) ·
[Contract-flow runs](#contract-flow-runs-company-side) ·
[Rate limits](#rate-limits) · [Errors](#errors) · [How it's wired](#how-its-wired)

Deeper reference pages live in [`docs/`](docs/):
[config](docs/config.md) · [model](docs/model.md) · [pump](docs/pump.md) ·
[webhooks](docs/webhooks.md) · [errors](docs/errors.md).

---

## TL;DR — fetch new updates

A system/Homebrew Python refuses a bare `pip install` (PEP 668) — install into a
virtualenv:

```bash
python3 -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install allus-company-data
```

Point a config.json at your service keys:

```json
{
  "api_url": "https://api.allme.fyi",
  "client_id": "svc_xxx",
  "client_secret": "xxx",
  "service_private_key": "/path/to/service.pem",
  "key_passphrase": "xxx",
  "cache_dir": "./allus-cache"
}
```

Drain everything new, handled one update at a time:

```python
from allus_company_data import Client

client = Client.from_config("config.json")

def handle(change):
    # one update at a time: event, person, slug, value, live, at
    print(change.event, change.person_id, change.slug, change.value,
          "live" if change.live else "snapshot", change.at)

client.process_changes(handle)   # returns when the feed is empty
```

`process_changes` pulls every pending change, decrypts it, and hands them to your
callback ONE BY ONE, acking each only after your code returns. Crash mid-batch?
The next run replays exactly what wasn't acked — nothing is lost, and the API
keeps no backlog of its own. Run it on a schedule (cron / systemd timer); there
is no daemon/follow mode by design. Connections, binary values, and webhooks are
documented below.

---

## Quickstart

Requires **Python ≥ 3.11**. A system/Homebrew Python refuses a bare `pip install`
(PEP 668) — install into a virtualenv:

```bash
python3 -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install allus-company-data
# or, working from a clone:    pip install -e '.[dev]'      # from the repo root
python -c "import allus_company_data; print(allus_company_data.__version__)"
```

### 1. Write a config file

A single JSON file holds everything. Any field can be overridden by an `ALLUS_*`
env var, so secrets needn't live in the file. **No SDK method ever takes a key,
passphrase, or secret as an argument** — they all come from here.

`allus.json`:

```json
{
  "api_url": "https://api.allme.fyi",
  "client_id": "svc_1a2b3c…",
  "client_secret": "…",
  "service_private_key": "./service-CRM.pem",
  "key_passphrase": "…",

  "account_private_key": "./account.pem",
  "account_passphrase": "…",

  "webhooks": {
    "wh_abc123": "hmac_secret_for_that_webhook"
  },

  "cache_dir": "./allus-cache",
  "format": "json"
}
```

| Field | Required | Meaning |
|-------|----------|---------|
| `api_url` | yes | API base, e.g. `https://api.allme.fyi`. |
| `client_id` / `client_secret` | yes | The registered `client_credentials` credentials for **one** service. |
| `service_private_key` | yes | Path to the OpenSSL-encrypted PKCS#8 PEM you downloaded from the portal. |
| `key_passphrase` | yes | Decrypts that PEM in memory at startup. |
| `account_private_key` / `account_passphrase` | only for `encrypt_payload` webhooks | The company **account** key, used to unwrap an encrypted webhook envelope. |
| `webhooks` / `webhook_secret` | webhook auth — HMAC (default) | Per-webhook HMAC secrets keyed by webhook id (matched via the `X-Allus-Webhook-Id` header). A single-webhook service can use a flat `"webhook_secret": "…"` instead of the map. |
| `webhook_bearer_token` | webhook auth — bearer | Verify `Authorization: Bearer <token>` deliveries. |
| `webhook_basic` | webhook auth — basic | `{"username","password"}` — verify HTTP Basic deliveries. |
| `webhook_header` | webhook auth — header | `{"name","value"}` — verify a custom-header delivery. |
| `webhook_auth_none` | webhook auth — none | `true` — explicit opt-out; `verifyWebhook` always passes (use only behind your own gateway). **Configure at most one** webhook auth method (two+ → `ConfigError`). |
| `cache_dir` | no (default `./allus-cache`) | Durable local buffer for the changes pump. Must be writable + durable. |
| `format` | no (default `json`) | Wire format `json` or `xml`. Invisible in the output. |

Env overrides use the `ALLUS_` prefix of the field name, e.g.
`ALLUS_CLIENT_SECRET`, `ALLUS_KEY_PASSPHRASE`, `ALLUS_ACCOUNT_PASSPHRASE`,
`ALLUS_WEBHOOK_SECRET`. A missing/invalid config (or an unreadable PEM / wrong
passphrase) raises `ConfigError` at construction — fail fast.

### 2. First call — list a connection's values

```python
from allus_company_data import Client

client = Client.from_config("allus.json")

# Iterate every connected person (lazy, auto-paged).
for conn in client.connections():
    print(conn.display_name, conn.person_id)
    for slug, val in conn.values.items():
        print(f"  {slug} = {val.value!r}  (live={val.live}, updated={val.updated_at})")
    break  # just the first one for the demo
```

Or fetch one connection by id:

```python
conn = client.connection("019xxxxxxxxxxxxxxxxxxxxxxxxx")
email = conn.values["work_email"].value        # "alice@acme.com"  (a str)
```

`client = Client.from_env()` builds the same client entirely from `ALLUS_*`
env vars (no file).

---

## Every call

`Client` is the only object you construct. Build it from config, then:

```python
Client.from_config(path, **kwargs) -> Client     # from a JSON file (env overrides secrets)
Client.from_env(**kwargs)          -> Client      # entirely from ALLUS_* env vars
```

`kwargs` are advanced/optional: `http` (an injected `HttpClient`), `logger` (a
`logging.Logger`), `sleep` (a `Callable[[float], None]`, for tests).

### `request_fields()`

```python
request_fields() -> list[RequestField]
```

Your request-field **definitions** — fetched once from
`GET /api/company-data/request-fields` and cached for the life of the client (it
types every value). Returns *your* request config, never the person's fields.

* **Params:** none.
* **Returns:** `list[RequestField]` — each `RequestField(slug, label, type, one_time, mandatory, raw)`. `mandatory` is true when the field is mandatory-to-provide **or** mandatory-to-stay-connected.
* **Raises:** `AuthError`, `ApiError`, `RateLimitError`.

```python
for f in client.request_fields():
    flag = "mandatory" if f.mandatory else "optional"
    print(f"{f.slug:20} {f.type:10} {flag}{' (one-time)' if f.one_time else ''}")
```

### `connections(limit, offset)`

```python
connections(limit: int = 100, offset: int = 0) -> Iterator[Connection]
```

A **lazy generator** that auto-pages `GET /api/company-data/connections?limit&offset`
and yields one typed `Connection` at a time (bounded memory for a large book).
Each `conn.values[slug]` is already decrypted (or a lazy binary handle).

* **Params:** `limit` — page size (default 100); `offset` — starting offset.
* **Returns:** `Iterator[Connection]`.
* **Raises:** `AuthError`, `ApiError`, `RateLimitError` (after the iterator's bounded internal backoff — see [Rate limits](#rate-limits)).
* **A value the service key cannot open never ends the listing.** It is returned in its own place with `value` `None` and `unreadable` `True` (see [`Value`](#value)), and every other value and connection is returned as usual. When every value of every connection reads `unreadable`, check the configured `service_private_key`.

> **Heavily rate-limited.** Use for the initial full sync + occasional
> reconciliation only — never as a poll substitute for the changes feed. The
> generator paces itself within the limit (backs off on `Retry-After`).

```python
# Initial full sync, streaming so a 100k-connection book never lands in memory.
for conn in client.connections(limit=200):
    upsert_local_record(conn)
```

### `connection(id)`

```python
connection(id: str) -> Connection
```

Fetch one connection by its connection id (`GET /api/company-data/connections/{id}`).

* **Params:** `id` — the connection id (`Connection.id`).
* **Returns:** one `Connection`. Note: this endpoint returns `{connection_id, user_id, values}` and **no** `display_name`/`connected_at`, so those identity fields are `None` here (the list endpoint carries them).
* **Raises:** `AuthError`, `ApiError` (404 if unknown), `RateLimitError`. A value the service key cannot open is returned marked `unreadable`, never raised.

```python
conn = client.connection(conn_id)
phone = conn.values.get("mobile")
if phone:
    print(phone.value, "live" if phone.live else "snapshot")
```

### `delete_connection(connection_id)`

```python
delete_connection(connection_id: str) -> None
```

End one of this service's connections (`DELETE /api/company-data/connections/{id}`). It leaves exactly the state the customer's own disconnect leaves: the link, the answers and copies held through it and the pair's messages on this service are deleted; the company connection, documents, signatures and flow runs stay. The customer is told by the platform, and `connection_deleted` reaches your change feed and webhooks. A customer whose account was deleted can still be disconnected. Reconnecting stays possible.

* **Returns:** nothing.
* **Throws:** `AuthError`, `ApiError`, `RateLimitError`; refusals are `404` `company_data.connection_not_found` (an id that is not a connection of this service) and `409` `company_connections.active_contract` (the customer holds an active agreement or subscription on this service).

```python
client.delete_connection(conn_id)
```

### `logs(limit, offset)`

```python
logs(limit: int = 50, offset: int = 0) -> list[LogEntry]
```

The service's activity log (`GET /api/company-data/logs?limit&offset`) — **ops
events only** (email / purge / webhook), never person field data.

* **Params:** `limit` (default 50), `offset` (default 0).
* **Returns:** `list[LogEntry]` — each `LogEntry(type, message, metadata, at, raw)`.
* **Raises:** `AuthError`, `ApiError`, `RateLimitError`.

```python
for entry in client.logs(limit=20):
    print(entry.at, entry.type, entry.message)
```

### `process_changes(handler, **options)`

```python
process_changes(handler: Callable[[Change], None], **options) -> None
```

The crash-safe changes pump: drains the feed through `handler` **one `Change` at
a time**, durably buffering each batch before delivery, with per-item ack and
retry → dead-letter → continue. Runs **until the feed is empty, then returns** —
there is **no follow/daemon mode** (you schedule re-runs yourself). Delivery is
**at-least-once**, so your handler **must be idempotent** (dedup on `Change.id`).
See [The changes pump](#the-changes-pump) for the full model.

* **Params:** `handler` — your callback; called with one `Change`. A return is an ack; an exception triggers retry.
* **Options** (keyword-only): `batch_size` (clamped to ≤ 500, default 100), `max_retries` (default 3), `on_error` (`"deadletter"` — default — or `"halt"`), `backoff` (`Callable[[int], float]`, attempt → seconds).
* **Returns:** `None` (when the feed is empty + the buffer is drained).
* **Raises:** `AuthError`, `ApiError`, `RateLimitError` (during a drain); `ValueError` (bad `on_error`); whatever the handler raises if `on_error="halt"` and retries are exhausted.

```python
def handle(change):
    if already_processed(change.id):      # idempotency — dedup on the stable id
        return
    if change.event == "field_updated":
        store(change.person_id, change.slug, change.value)
    elif change.event in ("connection_deleted", "field_deleted"):
        remove(change.person_id, change.slug)
    mark_processed(change.id)

client.process_changes(handle)            # returns when the feed is empty
```

> `logger` is **not** a `process_changes` option in this SDK — pass it once to
> the `Client` constructor (`Client.from_config("allus.json", logger=my_logger)`).

### Advanced changes primitives

```python
drain_batch(max: int = 100)                      -> list[Change]   # raw, UNBUFFERED — you own durability
dead_letters()                                   -> list[dict]      # the local dead-letter store
retry_dead_letters(handler, **options)           -> int             # re-drive dead-lettered events; returns count re-driven
```

* `drain_batch(max)` — fetches one batch (clamped ≤ 500) and returns the decrypted `Change`s directly. It does **not** persist anything, so a crash loses what the API already deleted. Prefer `process_changes` for safe consumption.
* `dead_letters()` — each dict is the stored (ciphertext) event plus a flattened `error` and `attempts`.
* `retry_dead_letters(handler, **options)` — same `max_retries` / `on_error` / `backoff` options as `process_changes`; on success a record is removed, on repeated failure it stays dead-lettered (or re-raises under `"halt"`). Dead letters are never re-fetched from the API — the local store is their only home.

```python
for dl in client.dead_letters():
    print("stuck:", dl["id"], dl["error"], "after", dl["attempts"], "attempts")

n = client.retry_dead_letters(handle)     # after you've fixed the bug
print(f"re-drove {n} dead letters")
```

### Key rotation — `key_rotated` and the public-key cache

Every client caches the RSA public keys it fetches: a person's key is immutable — until they
**rotate** it. A person learns of a rotation from a silent push; your service gets no pushes, so the
`key_rotated` change is your **only** signal. Without it a long-running worker keeps encrypting to
the rotated-away key for its whole lifetime, and the person can never read those values.

**On the pump this is automatic** — the cached key is dropped as the change passes through, before
your handler sees it. **Over a webhook it is not:** the signature verifier is static and has no
client instance, so it cannot reach the cache. Call the invalidator yourself — noting that the two
clients key their caches **differently**: the service client by `share_code`, the customer client by
the person's **user id**. Passing a share code to the customer client removes nothing and leaves you
encrypting to the old key. Both identifiers ride every change, alongside `public_key_sha256` — the
fingerprint of the person's new key.

```python
if change.event == "key_rotated":
    client.invalidate_public_key(change.share_code)      # service Client — keyed by SHARE CODE
    customer.invalidate_public_key(change.person_id)     # CustomerClient — keyed by PERSON USER ID
    # change.public_key_sha256 = fingerprint of the NEW key, if you want to verify the refetch
```

This is **eventual, not fail-closed** — nothing rejects a document encrypted to a stale key, so a
window remains between the rotation and your next drain. Drain often if that window matters.

### `service_key_rotated` — the same thing, the other way round

The customer client also caches the **service's** public key, the one you encrypt your consent
answers and documents *to*, keyed `"companyCode/serviceCode"`. When that company replaces its
service keypair, the `service_key_rotated` change on your account feed is your only signal — you
receive no pushes. Same shape, same guarantees, same automatic handling on the pump:

```python
if change.event == "service_key_rotated":
    # Automatic on the pump. Over a webhook, from the raw event body:
    customer.invalidate_service_key(body["company_share_code"], body["service_share_code"])
    # body["service_public_key_sha256"] = fingerprint of the service's NEW key
```

Also **eventual, not fail-closed**. Note the identifiers are **share codes**, not the ids used by
`invalidate_public_key` — the two caches are keyed differently and the wrong call removes nothing.

### Webhook helpers (on the client)

The webhook receiver helpers are also exposed as `Client` methods (they delegate
to the module functions, fully config-driven — no key/secret arguments):

```python
client.verify_webhook(raw_body: bytes, headers: dict) -> bool
client.parse_webhook(raw_body: bytes, headers: dict)  -> Change
client.handle_webhook(raw_body: bytes, headers: dict) -> Change   # verify + parse
```

* `verify_webhook` — recomputes `HMAC-SHA256(raw_body, secret)` and constant-time-compares it to `X-Allus-Signature`. Returns `True`/`False`; **never raises** for a bad signature.
* `parse_webhook` — body → a typed `Change`. Does **not** verify. Handles JSON, XML, and the `encrypt_payload` account-key envelope. Raises `WebhookError` on a malformed/unparseable body.
* `handle_webhook` — verify **then** parse; raises `WebhookError` on a bad/unknown signature, otherwise returns the `Change`. The typical one-liner inside a route.

The same three are importable as standalone functions
(`from allus_company_data import verify_webhook, parse_webhook, handle_webhook`),
which take the `config` and the decrypt/type closures explicitly — but inside an
app you'll almost always use the client methods. See [Webhooks](#webhooks).

---

## The typed value model

You work with these objects and nothing else (`from allus_company_data import …`):

```text
RequestField { slug, label, type, one_time, mandatory, verified, verified_max_age_days }
Connection   { id, person_id, display_name, connected_at, values: {<slug>: Value} }
Value        { value, live, updated_at, verified, verified_at, verified_expires_at,
               verified_method, verified_provider, verification_id, unreadable }
Change       { id, event, person_id, slug?, value?, live?, document_id?, status?, at,
               sealed_at?, plain_sha256?, signer_first_name?, signer_last_name?, signer_name_verified? }
Document     { id, kind, name, description, status, payload_kind, is_private, value, metadata,
               created_at, updated_at, plain_sha256, sealed_at, signatures }
LogEntry     { type, message, metadata, at }
```

### Keyed by *your* slug

`conn.values["work_email"].value` → `"alice@acme.com"`. The key is the stable,
explicit slug you set per request field in the portal — rename the label freely,
the slug is the contract. **The person's source field is never exposed**: no
source slug, no `field_id`, not even via `.raw`.

### `Value(value, live, updated_at)`

| Attribute | Meaning |
|-----------|---------|
| `value` | The typed plaintext (see the table below). |
| `live` | `True` if the person chose "keep connected" (auto-updates); `False` for a one-time snapshot. |
| `updated_at` | `datetime` of when this answer last changed (per-answer, rides on the `Value`). |
| `verified` | `True` only when the verification hash recomputes over the decrypted plaintext **and** the verification has not lapsed. Absent metadata reads `False`, which means "not attested", not "wrong". |
| `verified_at` | `datetime` the answering field was verified, or `None`. A stamp, not a promise about today. |
| `verified_expires_at` | `datetime` that verification lapses, or `None` when it does not. A document-backed verification dies with the document; once this is past, `verified` reads `False`. |
| `verified_method` | HOW allme bound the value: `email_code` \| `sms_code` \| `sumsub_id` \| `sumsub_address`. |
| `verified_provider` | WHO established the proof: `allme` \| `sumsub`. |
| `verification_id` | The proof id to quote back to allme in a dispute — it resolves the full record, including facts you never receive. |
| `unreadable` | `True` when the answer is present but the configured service key cannot open it — sealed to a key the service has since replaced, or a wrong configured key. `value` is then `None` and `verified` `False`; every other attribute is read as for a readable value. |

**Not readable is not empty.** An unanswered value is `value` `None` with `unreadable` `False`; a value that could not be opened is `value` `None` with `unreadable` `True`. A binary value is a lazy handle and is never marked: a binary whose file cannot be opened fails when its bytes are read. When every value of every connection reads `unreadable`, check the configured `service_private_key`.

`verified_method`, `verified_provider` and `verification_id` are the **proof metadata** and arrive **together or not at all**: a value bound before
the proof log existed carries the four verification keys and none of these, so all three read `None`.
They are readable whatever `verified` says — that boolean stays the only trust decision.

### Value types — from the type's RESOLVED definition

A contact-field TYPE is a ROW in the served field-type registry, not a name this SDK knows by
heart. The client fetches that registry (`GET /api/contact-field-types`) beside your request-field
catalog and holds it for its life; a value's shape follows the type's resolved storage LANE and
PRIMITIVE, so a type added as a row types itself with no SDK release.

| The type's resolved… | Python `value` |
|----------------------|----------------|
| storage lane `photo` / `document` | a lazy `BinaryHandle` — pages, declared entries and file bytes; see below |
| primitive `composite` | `dict` — the decrypted plaintext is a JSON object, parsed for you |
| primitive `date` | `datetime.date` (falls back to the raw string if it can't be parsed) |
| primitive `multilist` | `list` of the chosen option strings |
| the reserved type key `plugin` (typed first, before the registry) | a `PluginValue` — see below |
| anything else, and a type the registry does not carry | `str` |

For the seeded types that means, unchanged: `email`/`phone`/`url`/`text` → `str` (`phone` is a
single E.164-style string, `+` and digits); `country`/`nationality` → `str`, an ISO 3166-1 alpha-2
code (e.g. `"US"`, `"NL"`), not a display name;
`personal_number`/`place_of_birth` → `str`, and a verified place of birth is `<city>, <alpha-2>`
(e.g. `"Amsterdam, NL"`) — a country CODE, not a name; `address`/`bank`/`creditcard` → `dict`;
`date`/`date_of_birth` → `datetime.date`; `photo`, `document`, `legal_document` and the ID-document
subtypes `passport`, `photo_id`, `drivers_license` → a lazy `BinaryHandle`.

An `address`'s `country`/`state` sub-fields are an ISO alpha-2 code / USPS 2-letter state code
respectively. `client.field_types().is_field_value_valid(type, value)` validates a plaintext
against its type; `is_valid_country_code(code)` / `dial_code_for(code)` check a code or look up its
E.164 dial code.

Reach the registry itself for `resolve()`, `accepts(requested, actual)`, `descendants()`,
`is_binary()`, `label_for()`, `ordered()` and `validate(type, value)` (`None` when valid, else the
name of the first failing rule). A request row of a PARENT type MAY be answered by a field of any DESCENDANT — a `legal_document`
slot can be answered with a passport — but that matching is the API's and stays there. The answer
reaches you keyed by YOUR slug and typed by the SLOT's own type: the person's source field is
never exposed, so there is no source slug, no `field_id` and no source type to resolve, not even
via `.raw`. For a binary slot the API resolves slot → source → file itself, so the bytes you fetch
are the answering field's whatever type that field carries.

**A CHOICE type's options.** `select` and `multiselect` carry no options of their own — a flow
element supplies them — so `validate` / `is_field_value_valid` / `field_value_error` take an
optional third argument, the caller's own option list. The ROW's resolved options govern whenever
it has any, else the supplied list, and never a merge of the two; `client.field_types().options_for(type, options)`
answers exactly the domain the validator will enforce, or `None` when neither source has one. A
choice with no domain at all is refused (rule name `options_unavailable`) rather than measured
against an empty list. `submit_flow_answers` passes the flow element's options for you.


```python
addr = conn.values["home_address"].value     # dict, e.g. {"street": "...", "city": "...", ...}
dob  = conn.values["birthday"].value          # datetime.date(1990, 5, 17)
```

### Plugin rows — `RequestField.plugin` and `PluginValue`

A company can ask a request row through a **plugin** it added in the portal. Such a row's `type` is
the reserved key `plugin`, and its definition carries `plugin` —
`RequestFieldPlugin(plugin_name, field_type, snapshot)`, `snapshot` being the field type's frozen
description (`label`, `blocks`, `inputs`, `outputs`); every other row reads `plugin=None`. A plugin
row is always one-time and is asked of people only.

Its value is a `PluginValue` — the self-describing answer, readable without the plugin:

```python
PluginValue(
    plugin,    # the plugin's name, e.g. "Flex"
    type,      # the field type, e.g. "cao"
    blocks,    # [{key, kind, label, id, value}] in declared order; a search_select pick carries id + its option label as value
    outputs,   # [{key, type, label, value}] typed as the plugin declared them
    raw,
)

cao = conn.values["cao"].value
next(o["value"] for o in cao.outputs if o["key"] == "min_wage")   # 9.5
```

A plugin answer is what the person's client submitted — sealed to your service key but **not
signed**. When you must rely on an output, check it with the plugin itself.

### Binary fields — the lazy `BinaryHandle`

A photo/document value is a `BinaryHandle`. Nothing is fetched or decrypted until
you call `.bytes()` or `.save()`:

```python
handle = conn.values["passport_scan"].value   # BinaryHandle (no network yet)

pages = handle.pages()                         # GET the slot file → every page, in order
meta  = handle.metadata()                      # the entries the type declares
data  = handle.bytes()                         # the primary file bytes (single-file answers)
n     = handle.save("/tmp/contract.pdf")       # same, written to disk; returns bytes written
print(handle.value_url)                         # the opaque slot-keyed URL it fetches from
print(handle.content_type)                      # what the answer arrived as, once fetched
print(handle.content_sha256)                    # the platform's digest of the served artifact

for page in pages:
    print(page.label, page.name, page.mime, len(page.bytes))
print(meta.get("document_number"), meta.get("expiry_date"), meta.get("name"))
```

The slot-keyed file endpoint has **three 200 shapes, and which one you get is not
yours to choose**. It depends on whether the person's source field is private AND on
the type of the field they answered with; both can change, and nothing announces
either in advance:

* **private source** → `application/json`, `{"encrypted": true, "value": <wrapper>}`.
  The handle decrypts the wrapper with your service key into the JSON ENVELOPE.
* **non-private source whose type stores more than one file or declares metadata
  entries** (the ID-document subtypes and `legal_document`) → `application/json`,
  `{"encrypted": false, "value": "<envelope>"}` — the same envelope in the clear.
  Nothing is decrypted.
* **every other non-private source** → the file's own `Content-Type` (`image/jpeg`,
  `application/pdf`, …) and the body IS the file.

The envelope is `{"full": "data:…", "thumb": …}` for a photo, `{"file": "data:…", …}`
for a single-file document, and `{"pages": [{"label": …, "file": "data:…", …}], …}` for
a multi-page one, with every entry the type declares beside it.

The handle hides the difference. `.pages()` gives you the pages of a multi-page answer
(and `[]` for a single-file one); `.metadata()` gives you the declared entries
(`document_number`, `expiry_date`, `issuing_country`, `name`, …) and `{}` when the
type declares none; `.bytes()`/`.save()` give you the primary file. **On a multi-page
envelope `.bytes()`/`.save()` raise `DecryptError("multi-page envelope: use pages")`**
rather than handing back the front page as though it were the whole document.
`.metadata()` carries **no ordering guarantee** — read the envelope string yourself if
you need the declared order.

The raw-bytes shape is told apart on the response `Content-Type`, never by looking at
the body — a PDF that happened to start with a brace must not be mistaken for a
wrapper — and the two JSON shapes on the body's `encrypted` member. All four
accessors share ONE lazy fetch: whichever you call first performs it, and the result
is cached, so repeated calls don't re-fetch.

Every 200 carries `X-Allus-Content-Sha256`, the sha256 of the **served artifact** —
the raw bytes on the bytes shape, the served `value` string on either JSON shape;
`handle.content_sha256` is that header (and `handle.content_type` the Content-Type),
so you can record what you received and later show your archived copy has not
drifted. It is the platform's word, not a signature. There is no variant selection.

A frozen (share-once) answer is retained for 90 days. After that the endpoint returns
**410 `company_data.file_expired`**, which surfaces as an `ApiError` whose `details`
carry the answer's `content_sha256` and `expired_at` — your archived copy is then the
only one, and you can still prove what it is:

```python
try:
    data = handle.bytes()
except ApiError as e:
    if e.error_key == "company_data.file_expired":
        log(e.details["content_sha256"], e.details["expired_at"])
```

### `Change(id, event, person_id, slug?, value?, live?, at)`

A change-feed / webhook event.

| Attribute | Meaning |
|-----------|---------|
| `id` | **The stable server change-row id — your dedup key** (captured before the server delete). |
| `event` | `connection_created`, `connection_deleted`, `field_updated`, `field_deleted`, `consent_accepted`, `consent_declined`, `document_status_changed`, `message_received`. |
| `person_id` | The person the change is about (may be `None`). |
| `slug`, `value`, `live` | Present only on `field_updated`; `value` is typed exactly like `Value.value` (incl. a lazy `BinaryHandle` for binaries). Connection/consent/document events carry no slot/value. |
| `document_id`, `status` | Present only on `document_status_changed` — which document moved lifecycle state and to what (no slug/value). See [Company documents](#company-documents). |
| `connection_id`, `message_id`, `person_public_key`, `message_body` | Present only on `message_received` — a person messaged your service. `message_body` is the **decrypted** text. See [Messaging](#messaging). |
| `verified`, `verified_at`, `verified_expires_at` | Present on `field_updated`, with the same meaning as on `Value`. |
| `verified_method`, `verified_provider`, `verification_id` | The proof metadata, same meaning and same all-or-none rule as on `Value`. |
| `at` | `datetime` of the change. (There is no separate `updated_at` on a change.) |

### `.raw`

Every model carries `.raw` — the underlying *hardened* API dict — for debugging
or an edge case the SDK didn't model. It still never contains the person's source
field.

See [`docs/model.md`](docs/model.md) for the full reference.

---

## The changes pump

The changes feed is a server-side **drain-on-fetch queue**:
`GET /api/company-data/changes?limit=N` returns up to N events (default 100, max
500) **and deletes exactly those rows in the same transaction** — no
offset/cursor, and the API keeps no copy afterward. So consumption can't be a
plain list: a consumer crash mid-batch would lose events the API already deleted,
and a huge backlog must not materialize in memory. `process_changes` solves both.

**Per run, repeating until the feed is empty then returning:**

1. **Replay first.** Deliver any un-acked events already in the local buffer (from a previous crashed run), oldest-first.
2. **Drain.** When the buffer is empty, fetch one batch and **persist it to the durable file buffer (fsync) BEFORE handing anything out.** This is the backup the API no longer has.
3. **Deliver one-by-one.** For each buffered event, oldest-first: decrypt its value *at delivery* (never on disk), build the typed `Change`, call `handler`.
4. **Ack / retry / dead-letter.** On success, remove the event from the buffer (ack). On a handler error, retry with backoff up to `max_retries`; then either move it to the dead-letter store and continue (`on_error="deadletter"`, default — one poison event never wedges the stream) or stop and re-raise (`on_error="halt"`). A `DecryptError` on a buffered event (corrupt/truncated ciphertext, rotated key) is **dead-lettered immediately** — re-decrypting can't fix it, so it does *not* burn retries (under `on_error="halt"` it re-raises). Either way it never propagates out and wedges replay.
5. Repeat until a drain returns empty **and** the buffer is drained → return.

### The durable buffer

* Plain files under `cache_dir` (zero extra dependencies): `pending/` for un-acked events, `deadletter/` for ones that exhausted retries.
* Stored events keep their **ciphertext** value — **no plaintext PII is ever written to disk**. Decryption happens only at delivery.
* Writes are crash-safe (temp file → fsync → atomic rename → dir fsync). Files are named with a monotonic, zero-padded sequence so they replay oldest-first.

### Crash safety, at-least-once, and idempotency

A batch is durably buffered *before* any delivery, and acked per-item only *after*
the handler succeeds. The ack can't be atomic with your side-effects — a crash
between your handler's success and its ack re-delivers that event on the next run.
That makes delivery **at-least-once**, so:

> **Your handler must be idempotent. Dedup on `Change.id`.**

`Change.id` is the stable server change-row id, captured before the server delete,
so it survives crash + replay unchanged.

### No follow mode

`process_changes` returns when the feed empties. **You** schedule re-runs — a
cron job, a `while True: client.process_changes(handle); time.sleep(5)` loop, a
worker queue, whatever fits. The feed is cheap to poll (see
[Rate limits](#rate-limits)).

### Worked example

```python
import time
from allus_company_data import Client

client = Client.from_config("allus.json")

def handle(change):
    # Idempotent: skip anything we've already applied.
    if seen(change.id):
        return
    match change.event:
        case "field_updated":
            store_value(change.person_id, change.slug, change.value, live=change.live)
        case "field_deleted":
            clear_value(change.person_id, change.slug)
        case "connection_deleted":
            drop_person(change.person_id)
        case "connection_created" | "consent_accepted" | "consent_declined":
            note_event(change.person_id, change.event, change.at)
    record_seen(change.id)

# Schedule your own re-runs; process_changes itself returns when empty.
while True:
    client.process_changes(handle, batch_size=200, max_retries=5)
    time.sleep(5)
```

If a handler keeps failing, the event lands in the dead-letter store instead of
blocking the stream; inspect with `client.dead_letters()` and re-drive with
`client.retry_dead_letters(handle)` after fixing the cause. See
[`docs/pump.md`](docs/pump.md).

---

## Webhooks

Webhooks are the lower-latency push alternative to polling the changes feed. The
platform POSTs each change event to your configured webhook URL with:

* `X-Allus-Webhook-Id` — which webhook this is (selects the HMAC secret from config).
* `X-Allus-Signature` — `HMAC-SHA256(rawBody, secret)` as lowercase hex.
* the body — the same slug-keyed `Change` shape as the pull feed (JSON or XML).

All secrets/keys come from config; the helpers take **no key or secret
arguments**. Use the raw request body bytes (do not re-serialize a parsed body —
the HMAC is over the exact bytes the platform sent).

### Delivery contract — effectively unique, rarely replayed

Each queued event is POSTed **once**, and **only HTTP `200` counts as delivered** — a
`202`, a `204`, a 3xx redirect and every 4xx/5xx are all treated as a failure. On anything
other than `200` (or a timeout or connection error) the event is **not retried in place**:
it and the rest of the webhook's queue move to a durable server-side backlog and the
webhook is marked bad. The backlog is delivered later, either automatically when
the webhook next probes healthy, or when you drain it yourself with
`GET /api/company-data/changes?webhook_id=…` (delete-on-read).

So deliveries are **effectively unique** — with one rare exception. If your endpoint
processed an event but the platform never saw your `200` (your response timed out, or
you crashed after committing but before responding), the event is treated as failed
and replayed on recovery, so you receive it **again**. Nothing caps that at two: a
failed probe leaves its backlog row in place, so every later recovery attempt whose
`200` is likewise lost replays the same event once more. Inside that window the
contract is **at-least-once** — plan for one or more repeats, not for exactly one.

> **Do not use `change.id` as an idempotency key here.** On the webhook path the id is
> neither reliably stable nor reliably fresh, and a receiver cannot tell which one it is
> holding. A **live** delivery is built with no change row behind it, so its id is minted
> for that single POST — the later replay of the same event is rebuilt from a durable
> backlog row and therefore carries a **different** id. But a replayed delivery carries
> **that row's** id, and the row stays in place until it is delivered successfully, so a
> re-attempted replay arrives with the **same** id — which changes again if the event is
> re-backlogged after a further failure. An id check therefore misses the duplicate you
> are most likely to see and matches only a rarer one; it is not a contract. If you need
> strict idempotency, key on the **content** — event + person + slug/document + payload —
> never on the id.

**Webhooks and the pull feed are alternative integrations — consume one, never both.**
The id-dedup guidance in the changes-pump section above applies to the pump only, where
`change.id` is the real server change-row id.


### In a web route (Flask)

```python
from flask import Flask, request, abort
from allus_company_data import Client, WebhookError

app = Flask(__name__)
client = Client.from_config("allus.json")

@app.post("/allus/webhook")
def allus_webhook():
    try:
        change = client.handle_webhook(request.get_data(), dict(request.headers))
    except WebhookError:
        abort(401)              # bad / unknown signature, or unparseable envelope

    # Do NOT carry the pump's id-dedup over here: the webhook id is not an
    # idempotency key (see "Delivery contract" above). Key on content if you need one.
    apply_change(change)
    return ("", 200)   # 200 — the ONLY status allus counts as delivered
```

`verify_webhook` / `parse_webhook` let you split the steps if you prefer:

```python
if not client.verify_webhook(raw_body, headers):
    abort(401)
change = client.parse_webhook(raw_body, headers)
```

### Config-driven secrets

Per-webhook HMAC secrets live in the config `webhooks` map, keyed by webhook id;
the SDK reads `X-Allus-Webhook-Id` off the request and looks up the matching
secret. A single-webhook service can use the flat `"webhook_secret": "…"`
shortcut (or `ALLUS_WEBHOOK_SECRET`). An unknown/unconfigured id ⇒ verification
returns `False` (and `handle_webhook` raises `WebhookError`).

### The `encrypt_payload` account-key envelope

If a webhook has `encrypt_payload` enabled, the body is **replaced** by a
`{"_enc":1,…}` envelope encrypted to your company **account** key (and the HMAC is
over that envelope — the final bytes sent). `parse_webhook`/`handle_webhook`
unwrap it transparently using the configured `account_private_key` +
`account_passphrase`, then decrypt the inner field value with the service key — so
an encrypted-payload `Change` is identical to a plain one. If you receive such a
webhook without an `account_private_key` configured, you get a `WebhookError`.

> The account-key envelope uses OAEP-**SHA1** (OpenSSL's default), distinct from
> the OAEP-SHA256 used for person field values — the SDK handles this difference
> internally; you only supply the account key in config.

See [`docs/webhooks.md`](docs/webhooks.md).

---

## Company documents

Documents are content **your service issues to people** (a quote, a contract, a
JSON payload, a PDF) — the mirror image of the request slots. They come in two
shapes:

* **Broadcast** — no target. Sent to **every** connection on the service.
  **Plaintext** — you can't single-key-encrypt one value to all your
  connections, so a broadcast value is stored as-is.
* **Per-person** — targeted at one connection (`connection_id=` /
  `person_user_id=` / `share_code=`). **Automatically end-to-end encrypted to
  that recipient's public key** before it leaves the process. The server only
  ever stores ciphertext.

> **The encryption rule, plainly:** every **per-person** document is
> automatically end-to-end encrypted to the recipient's public key — for **any**
> `is_private` value. **Broadcast** documents are plaintext. `is_private` is a
> **device-display-only** flag (it picks lock-screen vs decrypt-on-load on the
> recipient's device), *not* what decides encryption — so `is_private=True` with
> **no** per-person target raises `ConfigError`. As everywhere in this SDK, **no
> method ever takes a key or secret argument** — the recipient key is fetched for
> you, and your own service key comes from config.

### Creating documents

`payload_kind` picks the body:

* `payload_kind="json"` — pass `json_value=` (a JSON-serializable object).
* `payload_kind="file"` — pass `file_bytes=` (`+ file_mime=`); the bytes are
  uploaded. For a **per-person** file the bytes are encrypted automatically too.

```python
from allus_company_data import Client

client = Client.from_config("allus.json")

# 1. BROADCAST plaintext json doc — every connection sees it, no target.
notice = client.create_document(
    kind="document",
    name="2026 price list",
    payload_kind="json",
    json_value={"plan": "pro", "monthly": 49, "currency": "EUR"},
)

# 2. PER-PERSON doc — auto-encrypted to that recipient's public key.
#    Target it by ANY one of connection_id / person_user_id / share_code.
contract = client.create_document(
    kind="document",
    name="Service agreement",
    payload_kind="json",
    is_private=True,                       # display-only; needs a per-person target
    connection_id="019xxxxxxxxxxxxxxxxxxxxxxxxx",
    json_value={"tier": "enterprise", "term_months": 12},
)

# A per-person FILE — the bytes are encrypted for the recipient automatically.
# plain_sha256 (SHA-256 of the raw PDF bytes) is computed for you when omitted;
# it is required by the server for a signable file document
# (requires_signature/requires_acceptance) and ignored on a json document.
with open("agreement.pdf", "rb") as fh:
    pdf_bytes = fh.read()
    pdf = client.create_document(
        kind="legal_document",
        name="Signed agreement",
        payload_kind="file",
        person_user_id="019yyyyyyyyyyyyyyyyyyyyyyyyy",
        file_bytes=pdf_bytes,
        file_mime="application/pdf",
        requires_signature=True,
        # plain_sha256=compute_plain_sha256(pdf_bytes),  # optional — computed for you otherwise
    )

# is_private without a target → ConfigError (a broadcast can't be locked).
```

### Listing, reading, updating, deleting

```python
list_documents(*, person_user_id=None, status=None, limit=100, offset=0) -> list[Document]
document(document_id)                                                     -> Document
document_file(document_id)                                                -> bytes    # #491: the file BYTES
flow_run_document(run_id, output_key)                                     -> bytes    # the run's OWN (service-key) copy of one output document
update_document_status(document_id, status)                              -> Document
update_document_metadata(document_id, *, metadata=None, name=None,
                         description=None)                               -> Document
delete_document(document_id)                                            -> None
```

```python
# All this service's documents (optionally filter by person and/or status).
for doc in client.list_documents(status="offering"):
    print(doc.id, doc.name, doc.status, doc.payload_kind, "private" if doc.is_private else "shared")

doc = client.document(contract.id)

# For a json doc, .json() returns the plaintext object — a per-person doc is
# decrypted with your service key on demand; a broadcast doc is already plaintext.
print(doc.json())

# #491: download a FILE document's bytes (metadata methods don't include them).
# A BROADCAST document is served plaintext and is returned as-is. A PER-PERSON /
# private document is encrypted to the RECIPIENT's key — the company cannot
# decrypt that, so this raises ApiError('documents.recipient_encrypted') instead
# of a doomed decrypt attempt.
try:
    pdf_bytes = client.document_file(contract.id)
except ApiError as e:
    if e.error_key == "documents.recipient_encrypted":
        # This is a per-person document; only the recipient can read it. For a
        # generated flow document, download the company's OWN copy of that output instead:
        pdf_bytes = client.flow_run_document(run.id, "out_1")
    else:
        raise

# Move it through its lifecycle / edit its metadata.
client.update_document_status(contract.id, "active")
client.update_document_metadata(contract.id, name="Service agreement (v2)",
                                metadata={"renewal": "auto"})

client.delete_document(notice.id)
```

A `Document` carries `id, kind, name, description, status, payload_kind,
is_private, value, metadata, created_at, updated_at, plain_sha256, sealed_at,
signatures` (and `.raw`). Use `.json()` on a `payload_kind="json"` document to
get the decrypted plaintext object.

A contract-flow-generated document can also read `status="waiting"` — a run-participant
copy whose signer has not been reached yet in the run's signing line. It is
read-only: `update_document_status` raises with `error_key="documents.run_managed"`
(409) if you try to write `status` on a run-participant document while it is `waiting`,
`ready_to_sign` or `offering` — that status moves only through flow generation, the
run's own advance, sign/accept, or a run cancel/decline. Such a document's
`run_signatures` carries the WHOLE run's signing line — one entry per (output document,
participant), in line order, each `{output_key, name, party_key, document_id, position,
status, action, acted_at}`; every document of the run carries the same summary.

### The document seal

Completing every required signature/acceptance on a signable document is not the
same as sealing it. When the last one is recorded the platform *attempts*, on that
same request, to append a Signatures page and sign the whole PDF with a platform
certificate, replacing every party's copy with the sealed PDF — the platform's own
attestation, layered over each signer's existing act. The attempt can fail (no PDF
bytes on the completing act, a byte mismatch, the sealing service unavailable, or a
custodian-completed ward act, which never carries bytes) without affecting the
signatures themselves or the document's completed status — it is simply left
unsealed, and any party can seal it afterwards from their own device or the owning
company's portal (no SDK call triggers a seal). `doc.sealed_at` is `None` until a
seal actually succeeds; `doc.plain_sha256` is the SHA-256 of the document's
unencrypted PDF bytes (`None` on a json document, and on a file document with no
stored plaintext hash — such a document cannot be sealed). Each entry of `doc.signatures`
additionally carries `plain_sha256` (the hash the signer's client computed over the
bytes it showed), `signer_first_name`, `signer_last_name` (the acting human's own
profile name) and `signer_name_verified` (`True` iff the submitted name matched that
account's verified ID name) beside its existing `action`, `method`, `content_sha256`,
`ip`, `user_agent`, `created_at` members.

### Reacting to a status change in the feed

When someone advances one of your documents (e.g. signs it), the platform emits a
**`document_status_changed`** change. In a `process_changes` handler it carries
`.document_id` and `.status` (and **no** `slug`/`value` — it's a lifecycle event,
not a field value). A transition to `active` additionally carries the same seal
state the document read carries — `.sealed_at`, `.plain_sha256`,
`.signer_first_name`, `.signer_last_name`, `.signer_name_verified` — so you never
need a follow-up `document(id)` call just to learn a run sealed:

```python
def handle(change):
    if change.event == "document_status_changed":
        # the document moved lifecycle state (offering → ready_to_sign → active → …)
        on_document_status(change.document_id, change.status)
    elif change.event == "field_updated":
        store(change.person_id, change.slug, change.value)

client.process_changes(handle)
```

---

## Messaging

Your service can hold a **conversation** with a connected person — the same
messaging surface the person already uses, with your service as the counterpart.
Two shapes:

* **1-on-1** — `send_message(connection_id, text)`. **End-to-end encrypted**: the
  SDK encrypts one copy to the person's public key and one to your service key
  before anything leaves the process, so the person reads it in their app and you
  can re-read your own outbound text. The platform stores ciphertext only.
* **Broadcast** — `broadcast_message(text)`. One **plaintext** message to every
  person connected to the service (one body cannot be single-key-encrypted to all
  of them, exactly as for a broadcast document). It seeds each person's ordinary
  1-on-1 thread; their reply comes back end-to-end encrypted.

`send_message` answers **201** with the created message carrying `message_id`, and
returns that id — the value you hand back as the acknowledgement boundary.

Inbound messages arrive on the **changes pump / webhook** as a `message_received`
event — a person→company message only. A broadcast raises no event of its own.

```python
def handle(change):
    if change.event != "message_received":
        return
    print(change.person_id, change.message_body)   # already decrypted for you

    # Reply on the same connection. person_public_key rides the event, so no
    # second key lookup is needed.
    client.send_message(
        change.connection_id,
        "Thanks — we're on it.",
        person_public_key=change.person_public_key,
    )

    # Acknowledge what you handled. REQUIRED: without it the message stays
    # unread forever, your unread count grows, and the person never sees a read
    # receipt. Sending a reply does NOT acknowledge anything.
    client.mark_messages_read(change.connection_id, change.message_id)

client.process_changes(handle)
```

`mark_messages_read` is bounded by the boundary message: a message that arrived
while you were working is **not** swept, and a repeat is a no-op. The boundary
must be a message the person sent on that connection — anything else is refused
with `ApiError("company_data.ack_boundary_invalid")` (400).

```python
# One plaintext announcement to everyone connected to the service.
client.broadcast_message("We're closed on Friday.")
```

Refusals surface as `ApiError` carrying the platform `error_key`:

| `error_key` | Status | Meaning |
|-------------|--------|---------|
| `messages.messaging_not_entitled` | 403 | The company's plan does not include messaging. |
| `messages.not_connected` | 403 | The person is not connected to this service. |
| `messages.messaging_suspended` | 403 | Messaging is suspended for this service (or the whole company). |
| `messages.broadcast_suspended` | 403 | Broadcast alone is suspended for this service. |
| `messages.encryption_required` | 400 | A 1-on-1 body was not a valid encrypted wrapper. |
| `messages.broadcast_audience_too_large` | 422 | The service has more connections than a broadcast allows. |
| `messages.rate_limited` | 429 | Too many 1-on-1 messages to the same person. |
| `company_data.ack_boundary_invalid` | 400 | The ack boundary is not a message the person sent on that connection. |

---

## Contract-flow runs (company side)

The company is one bound **party** of a contract flow — a multi-step, per-party
form the platform walks to gather (and end-to-end encrypt) answers, optionally
finishing at a document-generating leaf. These calls cover the company's turn:

```python
trigger_flow_run(flow_id, *, connection_id, bindings, source_files=None) -> FlowRun
stage_run_file(flow_id, source_user_id, sealed_value)      -> str      # staged file name
upload_answer_file(run_id, slug, for_user_id, sealed_value) -> str     # answer file name
flow_run_source_file(run_id, source_key)                   -> dict     # the sealed wrapper
published_flow(flow_id)                                    -> PublishedFlow
flow_runs(*, status="awaiting_company")                    -> list[FlowRun]
flow_run(run_id)                                            -> FlowRun
flow_run_answers(run)                                        -> FlowRunAnswers
submit_flow_answers(run, fill, *, party_pubkeys=None)       -> FlowRun
generate_flow_document(run)                                  -> dict
process_flow_run(run_id, fill_node, *, party_pubkeys=None)  -> FlowRun
identity()                                                    -> dict           # #491 gap 3
```

* `trigger_flow_run(flow_id, connection_id=..., bindings={...})` starts a run bound to a connection and the flow's other parties, pinning the flow's latest **published** version.
* `trigger_flow_run` reads the flow's latest published version first (`published_flow(flow_id)` — its `version`, `definition` and the service's request-field types) and sends it as `flow_version`. When that version's text elements show the customer's shared values (`{{party.field}}` tags), it opens those values from the connection with the service key and seals them per recipient — ONE wrapper of the non-private values, and each private value on its own — to the company (the service key) and to the customer, and sends them as `tag_values`; the connection's `values_private` decides which values are private (a slug it does not name is private). A newer publish in between (`flows.version_changed`) is re-read and retried once; a customer key that changed (`flows.tag_values_stale`) is re-read and retried once; a stale SERVICE key raises a `ConfigError` — rebuild the client with the service's current private key. The run carries `owner_tag_values` (the owning company's profile values the text names, plaintext) and `tag_values` (the company's sealed set).
* `flow_runs(status=...)` / `flow_run(run_id)` list / fetch runs. `status=None` returns everything; the default `"awaiting_company"` is the actionable queue.
* `flow_run_answers(run)` — a run's answers as a `FlowRunAnswers`, reading the company's service-key answer copies: `.answers` is the **decrypted** `{slug: plaintext}` map, `.unreadable` the list of slugs whose answer the service key could not open (empty when every answer opened). An unreadable answer is left out of `.answers` and never fails the call. Accepts a loaded `FlowRun` or a run id (fetched via `flow_run`).
* `submit_flow_answers` / `generate_flow_document` / `process_flow_run` fill the company's current node, advance the run (encrypting one answer copy per bound party), and — at a document-mode leaf — generate its output documents. See the method docstrings for the full per-party encryption details.
* `generate_flow_document(run)` returns `{"documents": [...], "status": ...}` — one `{output_key, party_key, document_id, position}` per produced (output document, participant). A leaf can produce several named output documents (e.g. "Contract" and "Addendum"); `position` is the step's 1-based place in the run's ONE signing line (one signer at a time, across every output), `None` for a party an output's signer list does not name. A repeat answers the same set.
* A `FlowRun`'s `participants` are `FlowRunParticipant(party_key, person_user_id, connection_id, documents)`; `documents` is that participant's own copy of each output document — `FlowRunParticipantDocument(output_key, name, document_id, document_status, requires_signature, requires_acceptance, position, action, acted_at)`, ordered by line position.
* `identity()` (#491 gap 3) — this client's own `{"company_user_id": ..., "service_id": ...}` from `GET /api/company-data/whoami`. `trigger_flow_run`'s company-side binding must use `company_user_id` (the person party's user_id comes from the connection) — without this call it was unconstructible through the SDK.

**Participant PDFs.** A leaf output rule's PDF is a company template, a flow field's answer
(`source_field`, source key `field:<slug>`, a field of type `pdf_document`) or what a bound customer
shared on its connection (`source_connection`, source key `conn:<party>:<request_slug>`). A rule
whose source the run does not hold does not match; the next rule is tried.

* **Connection sources are copied at run start.** For every answered `pdf_document` request slot a
  rule of the pinned version names, the caller reads the customer's shared file, seals its envelope
  JSON once per distinct bound user (the company's own copy to the service key) and stages each with
  `stage_run_file(flow_id, source_user_id, sealed_value)` (`POST /api/company-data/flows/{flowId}/run-files`)
  — `source_user_id` the customer bound to the source's party, whose shared PDF the copy is; the copy
  is stored in that customer's home region — then passes
  `source_files=[{"source_key", "for_user_id", "file"}]` to `trigger_flow_run`. A list that is not
  exactly the required set, or whose copy was staged for another customer than the one bound to its
  source's party, is refused with `ApiError` `flows.source_files_invalid`; its `details` carry
  `missing` (`[{source_key, for_user_id, source_user_id}]`) and `unexpected` (`[file]`), and nothing
  is written. Staging a copy for a customer that is not connected to the service is refused
  `flows.source_user_invalid`.
* `FlowRun.source_files` is the company's own copies, `{source_key: file}` (`{}` when none).
  `flow_run_source_file(run_id, source_key)` returns the stored copy — a wrapper that opens with the
  service key (`GET /api/company-data/flow-runs/{runId}/source-files/{sourceKey}`).
* **A company turn can answer a binary field.** `upload_answer_file(run_id, slug, for_user_id,
  sealed_value)` (`POST /api/company-data/flow-runs/{runId}/answer-files`) uploads one bound party's
  sealed copy; upload one per bound party, then submit `{"_enc_file": file}` as each party's value.
  An answer that is such a file reference reads as answered in the run's answer map.
* **Generation uploads the held source PDFs.** `generate_flow_document` (and `process_flow_run` at a
  document leaf) fetches the company's own copy of every source of the leaf's rules the run holds — a
  `source_field` whose own answer is a file (`slots/{slug}/file`), a `source_connection` in
  `source_files` — decrypts it, seals it under the call's one-time key and uploads it to
  `/generate/inputs` before generating with `inputs: [{source_key, input}]`. `flows.generate_inputs_mismatch`
  refuses inputs that are not the held set; `flows.source_pdf_invalid` refuses a source that is not a
  usable PDF (the run stays `generating`).

**The party that answers a run's last step generates the contract — the customer role included.**
When your company is a CUSTOMER of another company's service and its answer completes a document-mode
leaf, the run parks at `generating` until you generate:

```python
customer.generate_flow_document(connection_id, run)   -> dict   # POST /api/company-connections/{connection_id}/flow-runs/{run_id}/generate
```

Pass the run as re-read after your leaf submit. The answer map comes from your OWN copy of the run's
answers, decrypted with the account key — every party's answers are sealed to every bound party, so
that copy holds the whole run and no service key is involved. The held source PDFs are uploaded first
from your own copies (`answer-files`, decrypted with the account key), as on the service client. Returns `{documents, status}` — one
`{output_key, party_key, document_id, position}` per produced (output document, participant); a repeat
answers the same set. Raises `ConfigError` when the run's current step is
not bound to your company.

```python
me = client.identity()
run = client.trigger_flow_run(
    flow_id, connection_id=conn.id,
    bindings={"company": me["company_user_id"], "person": conn.person_id},
)

# Later, once the run is complete:
answers = client.flow_run_answers(run.id).answers   # {slug: plaintext} — the company's copies

# If the flow's output_mode is "document", download the company's OWN copy of every
# output document (encrypted to the SERVICE key, unlike a per-person document_file()):
run = client.flow_run(run.id)
own = next(p for p in run.participants if p.party_key == run.company_party_key)
for doc in own.documents:
    pdf_bytes = client.flow_run_document(run.id, doc.output_key)  # see Company documents above
```

* **Raises:** `AuthError`, `ApiError` (404 on `flow_run`/`flow_run_document` for an unknown run — `flows.run_not_found` — or, on `flow_run_document`, an output the run did not produce or a company that is not a bound party — `flows.no_document`), `DecryptError`, `RateLimitError`, `ValidationError` (from `submit_flow_answers` on a slug failing field-type validation, or a value outside its field's minimum/maximum).

### Plugin fields on a flow step

A flow step can hold a **plugin element** (`kind: "plugin"`): blocks (`search_select`, `text`,
`number`, `date`) answered by picks and typed values, inputs the company wired to earlier flow keys,
and outputs the plugin computes. When such a step is the company's turn, the service client talks
to the plugin through the platform's forwarder:

```python
plugin_pass(run_id)                                                          -> PluginPass
plugin_options(run_id, slug, block, query="", picks=None, values=None, draft=None) -> PluginOptions
plugin_outputs(run_id, slug, picks=None, values=None, draft=None)            -> PluginOutputs | PluginPicksInvalid
check_flow_value(run, slug, value, draft=None)                               -> None  # raises the bound ValidationError
```

* `plugin_pass` — `POST /api/company-data/flow-runs/{run_id}/plugin-pass` →
  `PluginPass(pass_, forwarder_url, plugins=[{"id", "public_key"}], specs={slug: spec})`. Issued only
  while the run awaits your party on that step; it lives ten minutes. The two calls below fetch one
  themselves.
* `plugin_options` — the options of one block: `query` is the search text (`""` lists everything),
  `picks` the ids picked so far by block key, `values` the typed block values so far →
  `PluginOptions(options=[{"id", "label"}], more)` (`more`: the list was cut — narrow the query).
* `plugin_outputs` — the outputs for `picks` and `values` → `PluginOutputs(outputs={key: value})`,
  or `PluginPicksInvalid` (`picks_invalid` is `True`) when the picks no longer fit the inputs or each
  other: clear them and pick again. **Call it again whenever an input changes**, before you submit.
* `draft` is the current step's not-yet-submitted answers (`slug → plaintext`). The plugin's inputs
  are read from ONE live answer map: the run's answers you can read, overlaid with `draft` for the
  current step's slugs, plugin answers expanded, constants computed. Each input is converted to its
  declared type (`number` → a JSON number, `date` → `YYYY-MM-DD`, `boolean` → a JSON boolean,
  `text` → a string).
* **Another party's private value is never sent to a plugin.** A source is private when its slug is
  in the run's `private_slugs`, when it is a constant that reads a private source, or when it is a
  current-step draft value whose field's `default` reads a private source (whatever the draft value
  is). A plugin answer's outputs are never private, whatever inputs produced them. A run
  read with no `private_slugs` list treats every other party's value as private. A REQUIRED input
  that is unwired, unanswered, private or not convertible raises `PluginInputUnavailable` (`input`,
  `source`, `reason`: `unwired` | `unanswered` | `other_party_private` | `not_convertible`); an
  OPTIONAL one is left out of the call.
* The call itself: the request is sealed to the plugin's public key with the platform wrapper and
  carries the public half of a fresh RSA-2048 reply key made for that call; it is posted to
  `{forwarder_url}/call` with a plain `requests.post` that carries **no allme credential** and uses
  the URL exactly as the pass names it; the reply is opened with the reply key. A
  `409 plugin.key_changed` reseals once with the key it returns; a 401 or 403 fetches a new pass
  once. Any other refusal is an `ApiError` carrying the forwarder's status and key
  (`plugin.not_responding`, `plugin.busy`, `plugin.rate_limited`, `plugin.unavailable`, …); a plugin
  that publishes no key reads `plugin.not_responding`.

The answer you then submit for the plugin slug is the plugin answer's JSON string:

```python
answer = json.dumps({
    "plugin": "Flex", "type": "cao",
    "blocks": [{"key": "cao", "kind": "search_select", "label": "CAO", "id": "hrc", "value": "Horeca Fictief"}],
    "outputs": [{"key": "min_wage", "type": "number", "label": "Minimum wage", "value": 9.5}],
})
client.submit_flow_answers(run, {"age": "19", "cao": answer, "wage": "14.20"})
```

Blocks go in declared order with their labels from the spec's `snapshot`; an answer without an
`outputs` array is unfinished and reads as unanswered.

**Defaults, minimums and maximums.** A flow field may carry `default`, `min` and `max`, each an
expression in the constants language (plugin outputs included, e.g.
`max(cao.min_wage, cao.agreement_wage)`). `submit_flow_answers` computes `min`/`max` over the same
live answer map (your `fill` as the draft) and refuses a value outside them with a `ValidationError`
naming the bound (`bound`: `"min"` | `"max"`, `bound_value`) before anything is encrypted; a bound
that computes to `None` is no bound. `check_flow_value(run, slug, value, draft)` applies the same
check on its own, before you submit. A value derived from another party's private source — a field whose
`default` reads one — is submitted marked
`source_private` (a run with no `private_slugs` list counts every other
party's value as private). Routing evaluates every edge over the answers with plugin answers
expanded, plus the computed constants.

**`FlowRun.private_slugs`** lists the slugs whose answers came from a private source — metadata,
never a value; `None` when the API did not send it (unknown).

**The customer role.** `CustomerClient` has the same three with a leading `connection_id`, over
`POST /api/company-connections/{connection_id}/flow-runs/{run_id}/plugin-pass`, reading its inputs
from its own copies of the run's answers (decrypted with the account key):

```python
customer.plugin_pass(connection_id, run_id)
customer.plugin_options(connection_id, run_id, slug, block, query, picks, values, draft)
customer.plugin_outputs(connection_id, run_id, slug, picks, values, draft)
customer.check_flow_value(run, slug, value, draft=None)   # raises the same bound ValidationError
```

Call `check_flow_value` before `encrypt_flow_answer` seals a value. `customer.submit_flow_answers`
reads the run once and marks every answer derived from another party's private source (the same
rule) `source_private: True` before it is sent.

The wire shape of each `answers[].values[].value` is the sealed wrapper serialized as a JSON string.
`submit_flow_answers` and `customer.submit_flow_answers` send it that way, whether `values[].value` is
the dict `encrypt_flow_answer` returns or a string; `encrypt_flow_answer` itself still returns the dict.

**The evaluator helpers** (exported, pure): `expand_plugin_answers(answers, plugin_slugs)` — a new
dict where each finished plugin answer becomes its summary (the blocks' values joined by `" / "`)
plus `slug.<block>`, `slug.<block>.id` (a `search_select` pick's id) and `slug.<output>`, and an
unfinished one is removed; `plugin_answer_summary(plaintext)`; `plugin_answer_view(plaintext)` →
`{"blocks": [{label, value}], "outputs": [{label, type, value}]}` or `None`;
`resolved_constants(constants, answers, reference_date, plugin_slugs=None)` expands first when given
the slugs. The math ops include variadic `max` and `min`, which skip what is not a number.

---

## Rate limits

| Endpoint | Limit | Use it for |
|----------|-------|-----------|
| `changes` (the pump) | **generous** | Poll **as often as you like** — it's a cheap drain-on-fetch queue. |
| `request-fields`, `logs` | moderate | Occasional reads. |
| `connections`, `connection(id)`, binary `/file` | **heavily limited** | Initial full sync + occasional reconciliation **only** — never as a poll substitute. |

A 429 carries `Retry-After`. The SDK backs off and retries automatically:

* The transport (`HttpClient`) retries a 429 a bounded number of times honoring `Retry-After`, then surfaces `RateLimitError`.
* The `connections(...)` generator additionally backs off per `Retry-After` on a surfaced `RateLimitError` and retries the page a bounded number of times before re-raising — so it paces itself within the limit instead of hammering.

If you catch a `RateLimitError`, its `.retry_after` is the seconds to wait
(or `None` when the header was absent).

Your `client_credentials` token requests (`/oauth2/token`) are on their own rate-limit bucket,
separate from person logins — but it is keyed by **source IP, not by your `client_id`**, so it is
shared with every other `client_credentials` caller reaching the API from the same address (another
service on your network, a second client on the same host). Caching the token, as described under
**How it's wired** below, is what keeps that shared window from being spent needlessly — by you or
anyone else behind the same IP. Every rate-limit refusal — this 429, and the platform's 503 when its
own limiter store is unreadable — now carries a populated `.error_key`, readable off the same
`RateLimitError`/`ApiError`.

---

## Errors

All from `allus_company_data`. Same taxonomy + names across all six SDKs.

| Error | When |
|-------|------|
| `ConfigError` | Missing/invalid config, unreadable key file, or wrong passphrase — at construction (fail fast). |
| `AuthError` | Token fetch/refresh failed (bad `client_id`/`secret`, revoked client); or a 401 survives the one automatic refresh-and-retry. |
| `ApiError(status, error_key, message, details)` | Any non-2xx from the API; carries the HTTP `status`, the platform `error_key` (when present), `message`, and `details` — the error body's remaining fields (e.g. a 410 `company_data.file_expired`'s `content_sha256` + `expired_at`). |
| `DecryptError` | A ciphertext wrapper is malformed, the key is wrong, or the GCM tag mismatches. Surfaces when a binary value's bytes are read, on a change event (the pump dead-letters it; a webhook parse raises it) and from flow-run routing and generation. `connections`/`connection` never raise it for a value — they mark it `unreadable` — and `flow_run_answers` lists such an answer under `unreadable`. |
| `WebhookError` | Signature verification failed, or an envelope couldn't be unwrapped/parsed. |
| `RateLimitError(retry_after)` | A 429 from a rate-limited endpoint. Subclass of `ApiError` (status fixed at 429); carries `retry_after` (seconds, or `None`). |
| `ValidationError` | A value failed its field type (`slug`, `field_type`), or a flow field's minimum/maximum (`bound`, `bound_value`). |
| `PluginInputUnavailable` | A required plugin input is unwired, unanswered, another party's private value, or not convertible (`input`, `source`, `reason`). |

```python
from allus_company_data import (
    Client, ConfigError, AuthError, ApiError,
    DecryptError, WebhookError, RateLimitError,
)

try:
    client = Client.from_config("allus.json")
    for conn in client.connections():
        ...
except ConfigError as e:
    ...   # fix the config / key file
except RateLimitError as e:
    wait(e.retry_after or 60)
except ApiError as e:
    log(e.status, e.error_key, e.message)
```

**One request waits 45 seconds for the platform's answer.** The SDK's own
transport — for `Client`, `CustomerClient` and `OAuthClient` alike — waits 45
seconds for the platform's answer to one request, and the call then fails as it
does when the connection drops; a `requests.Session` you pass in keeps its own
limit. A request given up may still have completed on the platform.

**A connection closed before the answer is tried once more.** When the platform
closes the connection without sending a single byte of the answer (`requests`
reports `RemoteDisconnected`), the SDK sends the request once more and reports
only the second failure; this also holds for a `requests.Session` you pass in. A
reset or a broken pipe is not resent. A request is never sent again once any
byte of its answer has arrived, after the 45 seconds ran out, or when the
connection could not be opened. The SDK cannot tell a reused connection from a
new one, so a first request on a new connection closed this way is sent again
too. A request the platform acted on before its connection died with no answer
at all runs twice.

**503 `db.writes_paused` — saving is paused, retry.** While the platform cannot
complete a save in every region, any call that is not a GET, the change-feed
drains (`process_changes`, `drain_batch`) and `OAuthClient.poll_result` can answer
`ApiError` with `status` 503 and `error_key` `db.writes_paused`, sent with
`Retry-After: 30`. Nothing was written (a drain drained nothing), so wait 30
seconds and repeat the same call; the SDK does not retry it. Reads and token
requests keep working.

**503 `platform.out_of_order` — the platform is out of order, retry.** While the
region serving a call is being rebuilt, every call — reads, the change-feed drains
(`process_changes`, `drain_batch`) and the `OAuthClient` calls included — can
answer `ApiError` with `status` 503 and `error_key` `platform.out_of_order`, sent
with `Retry-After: 300`. The request was not processed (a drain drained nothing),
so wait 300 seconds and repeat the same call; the SDK does not retry it. The
`client_credentials` token request keeps working; every other `/oauth2/token`
grant answers it.

See [`docs/errors.md`](docs/errors.md).

---

## How it's wired

Everything below is what the SDK hides so your code only ever sees conclusions.

**Auth / token.** An `HttpClient` owns a `client_credentials`-only token. On the
first call (or when the cached token nears expiry) it POSTs
`client_id`/`client_secret` to `{api_url}/oauth2/token` and caches the bearer
token + its expiry; refresh is automatic. A mid-flight 401 triggers exactly one
refresh-and-retry, then `AuthError`. The token is scoped server-side to **one**
service, so every call is implicitly that service's data.

**Regions.** The configured `api_url` is the platform's global front door and also the
starting point for every request, including the token request. A `client_credentials`
token is minted at your company's **home region**, and the token response names that
region's base in an `api_url` member — the SDK stores it and sends every subsequent
request there, token requests included, because the token is valid only at that region
and a company that moves region is followed by the next mint. A data call that still
reaches the front door is refused with `421` + `error_key: region.rebase_required`,
carrying the same `api_url`; the SDK stores it and retries the call exactly once. The SDK
does not validate a server-returned `api_url` against anything — it stores the base the
server names and uses it. An absent or empty `api_url` is never stored and the response
surfaces as the error it is.

**Slug resolution.** `request_fields()` is fetched once and cached; its slug→type
map types every value (so `address` parses to a dict, `photo` becomes a lazy
binary handle, etc.). A value or a change naming a slug that map does not carry —
a request slot configured after the client started — refetches the catalog ONCE,
and a slug still absent afterwards is remembered and never asked for again. The
connection/changes endpoints return values keyed by **your** request slug — the
person's source field is dropped server-side and never reaches the SDK.

**Decryption (zero-knowledge).** The service private key is loaded **once** at
construction from the configured encrypted PEM + passphrase into an in-memory RSA
key. A `decrypt` closure over it is handed to every model factory and the pump —
the key never appears in a method signature. Each value is a hybrid wrapper
(`{"_enc":1,"k":rsa_oaep_sha256(aesKey),"iv":…,"d":aes256gcm(…)}`); the SDK
RSA-OAEP-SHA256 unwraps the AES key, then AES-256-GCM decrypts the payload. **The
platform only ever holds ciphertext — it never sees your plaintext.**

**Binary fetch.** A binary value is a lazy `BinaryHandle` over a slot-keyed
`value_url`. On `.bytes()`/`.pages()`/`.metadata()`/`.save()` it GETs that file
endpoint once and classifies the response — the raw-bytes shape on its
`Content-Type`, the two JSON shapes on the body's `encrypted` member: an encrypted
answer (`{"encrypted":true,"value":…}`) runs the same service-key decrypt to the JSON
envelope, an `{"encrypted":false,"value":"<envelope>"}` answer carries that envelope
in the clear, and a plaintext answer's body already IS the file. Either way you get
the file, its pages and its declared entries, plus the response's
`X-Allus-Content-Sha256` on `.content_sha256`. (Slot-keyed, never
source-field-keyed.)

**The drain-on-fetch feed.** `process_changes` delegates to a `Pump` wired to a
`fetch_changes` closure (`GET /changes?limit=`, returning raw ciphertext events)
and a `decrypt` closure (builds a typed `Change`). Because the fetch deletes the
rows it returns, the pump persists each batch to the durable file buffer
(ciphertext at rest) before delivery, acks per-item after your handler succeeds,
and replays the buffer on restart — see [The changes pump](#the-changes-pump).

## Sign in with allme (OAuth, #195)

Relying-party helper for the "Sign in with allme" identity flow. Config-only keys (the idw role):

```python
from allus_company_data import OAuthClient, Claim

oauth = OAuthClient.from_config("idw-config.json")  # {api_url, oauth_client_id, oauth_redirect_uri, oauth_client_secret?, oauth_private_key?, oauth_key_passphrase?, authorize_url?}
url = oauth.authorize_url("signin", state="xyz", code_challenge=challenge)   # the button target
# ...user approves; your redirect_uri receives ?code=...
info = oauth.complete_sign_in(code, code_verifier=verifier)  # {user, mode, values(plaintext), values_cipher}
```

Modes: `signin` (identity), `one_time` (frozen claim values, decrypted for you), `connect` (a lasting connection),
`2fa_enroll` (opt a person into 2FA — see below).
`authorize_url(mode, claims=[Claim("email", "email", suggest="email_personal")])`; `poll_result(state)` for the detached response mode.

**The sign-in address** is the optional `authorize_url` of the same config (env `ALLUS_AUTHORIZE_URL`).
`authorize_url()` builds the button link on it, and on the live address `https://web.allme.fyi/auth` when it is absent.
The `authorize_url=` constructor keyword stays and wins over the config value.

**#498 — a claim IS a request field.** You describe what you need and the **person** picks which of their
own fields answers it; you never name a field. A claim carries a mandatory unique `name` (everything
that comes back is keyed by it — `values`, `values_cipher`, `attestations`, and their stored choice for a
repeat login), a field `type`, an optional `suggest`ed slug, `required`, and `verified` ("only a
#311-verified answer will do"). A nameless or duplicate claim raises `ConfigError` at the call rather than
failing at the API. `verified` is accepted only on the OIDC flow and only for a type allme can verify
(today `email`); elsewhere it is refused with `invalid_request` rather than quietly dropped.

`verified_max_age_days` narrows a `verified` claim to a RECENT verification, and the merge is **tighten-only**: the app's
registered configuration is a FLOOR, a request may only tighten it, and the effective limit is the minimum
of the two stated ages. An omitted age tightens nothing — omitting it sends nothing at all, never an
explicit null — and a value below 1 raises `ConfigError` at the call.

`complete_sign_in` returns `{user, mode, two_factor, values, values_cipher, attestations}`.
* `user.sub` **is** the person's share code and equals `share_code` — byte-identical to the id_token's
  `sub`. `display_name` is gone: ask for a `name` claim and read `values["name"]`.
* `values_cipher` is an additive sibling of `values`, keyed the same way: the raw app-key ciphertext
  wrapper each plaintext value was decrypted from, exactly as `userinfo` delivered it. Lets you show that
  a value really came from encrypted delivery rather than trusting it verbatim. Empty for a mode/claim
  that carries no ciphertext (`signin`, or `plaintext` delivery) — that emptiness is the honest answer.
* `attestations` is an additive sibling map keyed by the same claim name, present only for a `verified`
  claim under encrypted delivery. Each entry carries a `verified` boolean **the SDK computes itself**,
  in constant time, over the plaintext it just decrypted — plus the raw `hash`/`salt`/`verified_at`/
  `verified_expires_at`.
  **A slug ABSENT from the map is "not attested", never "wrong"** (treat that value as unverified);
  **an entry present with `verified` false is a MISMATCH and you must reject the value.** `verified_at`
  attests the value as verified *at that moment*, not verified today; `verified_expires_at` is when that
  verification lapses on its own (`None` = it does not), and an **expired attestation is unverified** —
  the computed `verified` already reads false once it has passed. `verified_method`,
  `verified_provider` and `verification_id` are the proof metadata read from the opened seal — HOW the
  value was bound, by WHOM, and the id to quote back to allme in a dispute. All three arrive together
  or not at all; a seal built before the proof log existed carries none of them and every one reads
  `None`.

**`resolve_userinfo(access_token, fallback_mode=None)`** is the second half of `complete_sign_in` — the
`userinfo` read + decrypt + attest, without the token exchange — for a caller whose exchange already ran
through a different client (a standards-only third-party OIDC library, say, that verified the id_token
itself but cannot read a claim value the id_token never carries). Config-only key handling applies exactly
as it does to `complete_sign_in`: you pass no key or passphrase, only the access token you already hold.
Returns the identical shape (`values`, `values_cipher`, `attestations`) and carries the same
mismatch-rejection duty on the caller. `complete_sign_in` is implemented on top of this method.
`fallback_mode` is used only when `userinfo` itself omits `mode` — pass the mode your own token
response carried, or `None` if you have none.


**Plugin claims.** An app may declare a plugin claim; its answer arrives in `values[name]` as the
plugin answer's JSON string. Read it with `parse_plugin_value(value)` → `PluginValue` (`plugin`,
`type`, `blocks`, `outputs`; a `ValidationError` when the value is not a finished plugin answer). A
plugin answer is always one-time — the consent screen asks it at every sign-in — and is sealed to
your app key but not signed: check an output with the plugin itself when you must rely on it.

## Building a plugin

A plugin is your own HTTPS service: `GET {base_url}/manifest` describes it (with your RSA-2048
public key) and `POST {base_url}/call` answers sealed requests. The SDK carries the sealing for
your server — functions for the plugin's server, never calls on the allme API:

```python
from allus_company_data import plugin_open_request, plugin_seal_reply

# POST /call — body {"request": "<wrapper string>"}
req = plugin_open_request(raw_body, open("plugin-key.pem", "rb").read(), None)   # unencrypted PKCS#8, or pass its passphrase
# req = {field_type, op: "options"|"outputs", block?, query?, picks, values, inputs, reply_key}
return plugin_seal_reply({"options": [{"id": "hrc", "label": "Horeca Fictief"}], "more": False}, req["reply_key"])
```

`load_private_key(pem, passphrase)` accepts an unencrypted PKCS#8 PEM with a `None` passphrase.
`generate_reply_key_pair()` makes an RSA-2048 pair (`(private_key, public_key_spki_b64)`) and
`export_public_key_spki(key)` exports a public key as base64 SPKI. A request sealed to a key your
plugin no longer holds should be answered `409 {"error": "key_unknown"}`.

## 2FA by allme (#436, #481)

Ask a connected person to approve a login inside the allme app. On the same service data client (no new
config), via the `two_factor` sub-client:

```python
from allus_company_data import Client

client = Client.from_config("allus.json")

# Raise a challenge. idempotency_key is REQUIRED — a repeat within the TTL returns the SAME challenge and
# sends no second push. `context` is plain text shown on the person's card.
ch = client.two_factor.challenge("2I6UF3", idempotency_key="login-8f3c1a", context="Sign-in from Chrome")
if ch.matching_digits:                        # number matching is on for this service
    show_on_login_page(ch.matching_digits)    # the person types these back into the app; the server checks them

# Wait for the terminal outcome — polls result() for you, raises ApiError on timeout.
res = client.two_factor.wait_for_result(ch.challenge_id)   # or result(ch.challenge_id) to poll once yourself
if res.status == "approved":
    grant_login()
```

- **Burn-on-read.** The first read of a terminal state (`approved` | `denied` | `expired` | `revoked`)
  delivers it and burns it — a later read is `gone`. Read it once and persist your own outcome;
  `wait_for_result` returns that first terminal read and never re-reads a consumed challenge.
- **Webhook variant.** The `2fa_challenge_completed` change/webhook carries the same terminal `status`, so a
  webhook consumer need not poll. **Expiry fires no webhook/Change** — only `approved`/`denied`/`revoked`
  reach the feed, so a lapsed challenge is observable only by polling.
- **Enrollment.** Only an enrolled person can be challenged (an un-enrolled `share_code` is `404`).
  Enrollment is a one-time consent on the `web.allme.fyi/auth` surface via the OAuth helper's `2fa_enroll`
  mode — a redirect button (`oauth.authorize_url("2fa_enroll", state=...)`), or server-to-server with
  `response_mode="detached"` + `poll_result(state)`, which returns `{"enrolled": true, "state": ...}` once
  the person confirms.
- **Errors.** `404` (unknown / not-enrolled share code). A `429` is either the plain rate limit (retried with
  backoff → `RateLimitError`) or `twofa.pending_cap` (too many challenges already open for this person) — the
  latter surfaces immediately as `ApiError` and is never retried, since a retry cannot clear it.
