From 180ecd5c39056934eccbb657e47f37103c86dc9a Mon Sep 17 00:00:00 2001 From: Waleed Qureshi Date: Wed, 21 Oct 2026 21:56:33 +0500 Subject: [PATCH] halvorai-ingest 0.2.0: validate before signing, env-aware endpoint, per-event results Fixes #1, #2 and #3. - emit validates every event against APE v0.1 (after redaction, as the Node.js SDK does); an invalid event is dropped, counted in stats().dropped_invalid and reported through on_error with the problems. It used to be signed and sent. - The default endpoint follows the key: hk_test_ keys go to the sandbox, hk_live_ keys to production. endpoint is a base URL like the Node.js SDK (a full /v1/events URL still works); HALVOR_API_KEY, HALVOR_SOURCE_KEY and HALVOR_ENDPOINT are read from the environment. - Delivery counts each event by its API status (accepted, duplicates, rejected) and reports each rejected event with its serverCode; events not sent for lack of a key count as dropped_disabled. Built from the monorepo with publish:build; not yet published to PyPI. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 9 + halvorai_ingest/client.py | 110 +++++++-- halvorai_ingest/validate.py | 363 ++++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_client.py | 7 +- tests/test_quickstart.py | 2 +- tests/test_validate_and_delivery.py | 171 +++++++++++++ uv.lock | 2 +- 8 files changed, 644 insertions(+), 22 deletions(-) create mode 100644 halvorai_ingest/validate.py create mode 100644 tests/test_validate_and_delivery.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e370c07..2a6743c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## 0.2.0 (unreleased) + +- `emit` validates every event against APE v0.1 before signing; an invalid event is dropped, counted in `stats().dropped_invalid` and reported through `on_error` with `invalid_event` and the problems found, as in the Node.js SDK. Previously it was signed and sent. +- The default endpoint follows the key: `hk_test_` keys (and `environment="test"` with no key) go to the sandbox, `hk_live_` keys to production. Previously every key went to production. +- `endpoint` is now a base URL, as in the Node.js SDK (`https://api.sandbox.halvorai.com`); a full `/v1/events` URL still works. `HALVOR_ENDPOINT` sets it from the environment. +- `HalvorClient()` reads `HALVOR_API_KEY` and `HALVOR_SOURCE_KEY` from the environment when they are not passed. +- Delivery counts each event by the status the API returned: `stats()` has `accepted`, `duplicates` and `rejected`, and each rejected event is reported through `on_error` with its `serverCode`. Events that could not be sent for lack of an API key count as `dropped_disabled`. +- `transport` accepts an `httpx` transport, for tests. + ## 0.1.0 (developer preview) - `HalvorClient`: batched, signed (EdDSA) event delivery; `emit` never raises. diff --git a/halvorai_ingest/client.py b/halvorai_ingest/client.py index b2b71a9..7fd8a67 100644 --- a/halvorai_ingest/client.py +++ b/halvorai_ingest/client.py @@ -3,11 +3,11 @@ from __future__ import annotations import json +import os import threading from copy import deepcopy from dataclasses import dataclass, field from typing import Any, Callable, Mapping -from urllib.parse import urljoin import httpx @@ -16,10 +16,25 @@ from .ids import new_event_id, now_utc from .redact import redact_deep, redact_text from .sign import sign_event +from .validate import validate_event -SDK_NAME = "halvorai-ingest@0.1.0" +SDK_NAME = "halvorai-ingest@0.2.0" DEFAULT_BATCH = 100 -DEFAULT_ENDPOINT = "https://api.halvorai.com/v1/events" +PRODUCTION_ENDPOINT = "https://api.halvorai.com" +SANDBOX_ENDPOINT = "https://api.sandbox.halvorai.com" + + +def default_endpoint(api_key: str | None, environment: str) -> str: + """Live keys (or environment='live' without a key) go to production; test keys to the sandbox.""" + if api_key is not None: + return PRODUCTION_ENDPOINT if api_key.startswith("hk_live_") else SANDBOX_ENDPOINT + return PRODUCTION_ENDPOINT if environment == "live" else SANDBOX_ENDPOINT + + +def events_url(endpoint: str) -> str: + """`endpoint` is a base URL, as in the TypeScript SDK; a full `/v1/events` URL is kept as is.""" + base = endpoint.rstrip("/") + return base if base.endswith("/v1/events") else base + "/v1/events" @dataclass @@ -27,8 +42,10 @@ class ClientStats: emitted: int = 0 queued: int = 0 accepted: int = 0 + duplicates: int = 0 rejected: int = 0 dropped_invalid: int = 0 + dropped_disabled: int = 0 dropped_closed: int = 0 @@ -40,10 +57,13 @@ class HalvorClient: api_key: str | None = None source_key: str | Mapping[str, Any] | None = None environment: str = "test" - endpoint: str = DEFAULT_ENDPOINT + #: Base URL of the API. Defaults to HALVOR_ENDPOINT, else sandbox for test keys, production for live. + endpoint: str | None = None sink: SinkFn | None = None batch_size: int = DEFAULT_BATCH on_error: Callable[[dict[str, Any]], None] | None = None + #: For tests: an httpx transport (e.g. httpx.MockTransport). + transport: httpx.BaseTransport | None = None _queue: list[dict[str, Any]] = field(default_factory=list, init=False, repr=False) _lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) @@ -53,6 +73,17 @@ class HalvorClient: _source_id: str = field(default="src_unknown", init=False, repr=False) def __post_init__(self) -> None: + # Like the TypeScript SDK, read keys from the environment (`init` writes them to .env.local). + if self.api_key is None: + self.api_key = os.environ.get("HALVOR_API_KEY") or None + if self.source_key is None: + self.source_key = os.environ.get("HALVOR_SOURCE_KEY") or None + if self.endpoint is None: + self.endpoint = os.environ.get("HALVOR_ENDPOINT") or default_endpoint( + self.api_key, self.environment + ) + if self.api_key is not None and self.environment == "test" and self.api_key.startswith("hk_live_"): + self.environment = "live" self._source_jwk = self._parse_source_key(self.source_key) if self._source_jwk is not None: kid = self._source_jwk.get("kid") @@ -107,6 +138,19 @@ def emit(self, input_event: Any) -> str: ctx = current_context() built = self._build_event(cloned, ctx, event_id) self._redact_event(built) + problems = validate_event(built) + if problems: + self._stats.emitted += 1 + self._stats.dropped_invalid += 1 + self.notify( + { + "code": "invalid_event", + "message": "event failed schema validation", + "eventId": event_id, + "details": problems, + } + ) + return event_id with self._lock: self._queue.append(built) self._stats.emitted += 1 @@ -247,13 +291,12 @@ def flush(self) -> dict[str, bool]: def _deliver_http(self, envelopes: list[dict[str, Any]]) -> dict[str, bool]: if not self.api_key: + self._stats.dropped_disabled += len(envelopes) self.notify({"code": "config", "message": "api_key required for HTTP delivery"}) return {"drained": False} - url = self.endpoint - if not url.endswith("/events") and not url.rstrip("/").endswith("events"): - url = urljoin(url.rstrip("/") + "/", "events") + url = events_url(self.endpoint or default_endpoint(self.api_key, self.environment)) try: - with httpx.Client(timeout=30.0) as client: + with httpx.Client(timeout=30.0, transport=self.transport) as client: res = client.post( url, json={"events": envelopes}, @@ -262,22 +305,57 @@ def _deliver_http(self, envelopes: list[dict[str, Any]]) -> dict[str, bool]: "Content-Type": "application/json", }, ) - if res.status_code < 300: - self._stats.accepted += len(envelopes) - return {"drained": True} + except Exception as exc: + self.notify({"code": "delivery_failed", "message": str(exc) or "http failed"}) + return {"drained": False} + if res.status_code >= 300: self._stats.rejected += len(envelopes) self.notify( { - "code": "rejected", + "code": "unauthorized" if res.status_code in (401, 403) else "rejected", "message": f"HTTP {res.status_code}", - "serverCode": str(res.status_code), + "serverCode": _error_code(res) or str(res.status_code), } ) return {"drained": False} - except Exception as exc: - self.notify({"code": "delivery_failed", "message": str(exc) or "http failed"}) - return {"drained": False} + self._apply_results(envelopes, res) + return {"drained": True} + + def _apply_results(self, envelopes: list[dict[str, Any]], res: httpx.Response) -> None: + """Count each event by the status the API gave it, as the TypeScript SDK does.""" + try: + results = res.json().get("results", []) + except Exception: + results = [] + by_id = {r.get("id"): r for r in results if isinstance(r, dict)} + for envelope in envelopes: + event_id = envelope["event"].get("id") + result = by_id.get(event_id, {}) + status = result.get("status", "accepted") + if status == "duplicate": + self._stats.duplicates += 1 + elif status == "rejected": + self._stats.rejected += 1 + error: dict[str, Any] = { + "code": "rejected", + "message": "server rejected event", + "eventId": event_id, + } + server = result.get("error") + if isinstance(server, dict) and isinstance(server.get("code"), str): + error["serverCode"] = server["code"] + self.notify(error) + else: + self._stats.accepted += 1 def close(self) -> None: self._closed = True self.flush() + + +def _error_code(res: httpx.Response) -> str | None: + try: + code = res.json().get("error", {}).get("code") + except Exception: + return None + return code if isinstance(code, str) else None diff --git a/halvorai_ingest/validate.py b/halvorai_ingest/validate.py new file mode 100644 index 0000000..44ffece --- /dev/null +++ b/halvorai_ingest/validate.py @@ -0,0 +1,363 @@ +"""APE v0.1 validation before signing, so a malformed event is dropped and reported, never sent. + +Mirrors the TypeScript SDK's zod model (`packages/schema/src/ape.ts`), which matches the published +JSON Schema at https://schema.halvorai.com/ape/v0.1.json, plus the card-number guard. +""" + +from __future__ import annotations + +import re +from datetime import datetime +from typing import Annotated, Any, Literal, Mapping, Union + +from pydantic import ( + AfterValidator, + BaseModel, + ConfigDict, + Field, + StringConstraints, + TypeAdapter, + ValidationError, + model_validator, +) + + +def _rfc3339(value: str) -> str: + if not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:\d{2})", value): + raise ValueError("must be an RFC 3339 date-time with an offset") + datetime.fromisoformat(value.replace("Z", "+00:00")) + return value + + +def _url(value: str) -> str: + if not re.fullmatch(r"[a-zA-Z][a-zA-Z0-9+.-]*://\S+", value): + raise ValueError("must be a URL") + return value + + +def _max(n: int) -> Any: + return StringConstraints(strict=True, max_length=n) + + +S16 = Annotated[str, _max(16)] +S32 = Annotated[str, _max(32)] +S64 = Annotated[str, _max(64)] +S128 = Annotated[str, _max(128)] +S256 = Annotated[str, _max(256)] +S500 = Annotated[str, _max(500)] +EventId = Annotated[str, StringConstraints(strict=True, min_length=8, max_length=128)] +DateTime = Annotated[str, StringConstraints(strict=True), AfterValidator(_rfc3339)] +Url = Annotated[str, StringConstraints(strict=True), AfterValidator(_url)] +Country = Annotated[str, StringConstraints(strict=True, pattern=r"^[A-Z]{2}$")] +Caip2 = Annotated[str, StringConstraints(strict=True, pattern=r"^[-a-z0-9]{3,8}:[-_a-zA-Z0-9]{1,32}$")] +Sha256 = Annotated[str, StringConstraints(strict=True, pattern=r"^sha256:[0-9a-f]{64}$")] +AmountValue = Annotated[str, StringConstraints(strict=True, pattern=r"^(0|[1-9][0-9]*)$")] +Currency = Annotated[str, StringConstraints(strict=True, pattern=r"^[A-Z]{3}$")] +FourDigits = Annotated[str, StringConstraints(strict=True, pattern=r"^[0-9]{4}$")] +RefValue = Union[Annotated[str, StringConstraints(strict=True)], Annotated[int, Field(strict=True)]] + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + + +class Source(_Strict): + id: S64 + sdk: S64 + adapter: Literal[ + "stripe", "stripe-issuing", "x402-client", "x402-server", "mpp-server", "mcp", "manual", "custom" + ] + + +class Kya(_Strict): + provider: S64 + subject: S256 + assurance: S64 | None = None + verified_at: DateTime | None = None + + +class Agent(_Strict): + id: S128 + name: S128 | None = None + framework: S64 | None = None + version: S64 | None = None + key_thumbprint: S64 | None = None + kya: Kya | None = None + + +class Owner(_Strict): + id: S128 + type: Literal["person", "business"] + display_name: S128 | None = None + country: Country | None = None + kyc_ref: S128 | None = None + + +class Instruction(_Strict): + id: S128 + hash: Sha256 + summary: S500 | None = None + channel: Literal["chat", "api", "schedule", "email", "agent", "other"] | None = None + issued_at: DateTime | None = None + tool_call_id: S128 | None = None + + +class Asset(_Strict): + network: Caip2 + address: S128 + symbol: S16 | None = None + decimals: Annotated[int, Field(strict=True, ge=0, le=36)] | None = None + + +class MinorAmount(_Strict): + value: AmountValue + unit: Literal["minor"] + currency: Currency + asset: Asset | None = None + + +class AtomicAmount(_Strict): + value: AmountValue + unit: Literal["atomic"] + asset: Asset + currency: Currency | None = None + + +Amount = Annotated[Union[MinorAmount, AtomicAmount], Field(discriminator="unit")] +PaymentStatus = Literal["requested", "authorized", "settled", "failed", "refunded", "disputed"] + + +class _PaymentBase(_Strict): + payment_id: S256 + amount: Amount + status: PaymentStatus + failure_reason: S128 | None = None + description: S256 | None = None + category: S64 | None = None + + @model_validator(mode="after") + def _failure_reason(self) -> "_PaymentBase": + if self.status == "failed" and self.failure_reason is None: + raise ValueError("failure_reason is required when status is failed") + return self + + +class StripeRef(_Strict): + account: RefValue | None = None + payment_intent: RefValue | None = None + charge: RefValue | None = None + refund: RefValue | None = None + dispute: RefValue | None = None + shared_payment_token: RefValue | None = None + + +class CardRef(_Strict): + authorization_id: RefValue | None = None + transaction_id: RefValue | None = None + network_transaction_id: RefValue | None = None + processor: RefValue | None = None + + +class X402Ref(_Strict): + scheme: RefValue | None = None + network: RefValue | None = None + nonce: RefValue | None = None + tx_hash: RefValue | None = None + facilitator: RefValue | None = None + payment_flow: RefValue | None = None + + +class MppRef(_Strict): + method: RefValue | None = None + challenge_id: RefValue | None = None + payment_intent: RefValue | None = None + tx_hash: RefValue | None = None + + +class StablecoinRef(_Strict): + network: RefValue | None = None + tx_hash: RefValue | None = None + log_index: RefValue | None = None + + +class BankTransferRef(_Strict): + scheme: RefValue | None = None + end_to_end_id: RefValue | None = None + pisp: RefValue | None = None + + +class StripePayment(_PaymentBase): + rail: Literal["stripe"] + rail_ref: StripeRef + + +class CardPayment(_PaymentBase): + rail: Literal["card"] + rail_ref: CardRef + + +class X402Payment(_PaymentBase): + rail: Literal["x402"] + rail_ref: X402Ref + + +class MppPayment(_PaymentBase): + rail: Literal["mpp"] + rail_ref: MppRef + + +class StablecoinPayment(_PaymentBase): + rail: Literal["stablecoin"] + rail_ref: StablecoinRef + + +class BankTransferPayment(_PaymentBase): + rail: Literal["bank_transfer"] + rail_ref: BankTransferRef + + +Payment = Annotated[ + Union[StripePayment, CardPayment, X402Payment, MppPayment, StablecoinPayment, BankTransferPayment], + Field(discriminator="rail"), +] + + +class Wallet(_Strict): + network: Caip2 + address: S128 + + +class Payee(_Strict): + type: Literal["merchant", "api_provider", "wallet", "business", "person"] + name: S256 | None = None + id: S128 | None = None + website: Url | None = None + country: Country | None = None + mcc: FourDigits | None = None + wallet: Wallet | None = None + resource_url: Url | None = None + + @model_validator(mode="after") + def _identified(self) -> "Payee": + if self.name is None and self.wallet is None and self.id is None: + raise ValueError("payee needs at least one of name, wallet or id") + return self + + +class Instrument(_Strict): + type: Literal["card", "agentic_token", "stripe_spt", "wallet", "bank_account"] + network: S32 | None = None + last4: FourDigits | None = None + par: S64 | None = None + token_ref: S128 | None = None + wallet: Wallet | None = None + issuer: S128 | None = None + + +class Evidence(_Strict): + kind: Literal[ + "ap2.payment_mandate", + "ap2.checkout_mandate", + "vi.chain", + "tap.signature", + "x402.payment_payload", + "x402.settlement", + "mpp.receipt", + "stripe.object", + "agentic_token.metadata", + "kya.attestation", + ] + format: Literal["sd-jwt", "kb-sd-jwt", "jws", "http-signature", "json"] + value: Union[Annotated[str, StringConstraints(strict=True)], dict[str, Any]] | None = None + uri: Url | None = None + sha256: Sha256 + captured_at: DateTime | None = None + + @model_validator(mode="after") + def _one_of(self) -> "Evidence": + if (self.value is None) == (self.uri is None): + raise ValueError("evidence needs exactly one of value or uri") + return self + + +class Related(_Strict): + previous_event_id: S128 | None = None + corrects_event_id: S128 | None = None + session_id: S128 | None = None + + +Labels = Annotated[dict[S64, S256], Field(max_length=20)] + + +def contains_pan(text: str) -> bool: + """True when text holds a Luhn-valid run of 13 to 19 digits (spaces or dashes allowed).""" + for run in re.findall(r"\d(?:[ -]?\d)*", text): + digits = re.sub(r"[ -]", "", run) + if 13 <= len(digits) <= 19 and _luhn(digits): + return True + return False + + +def _luhn(digits: str) -> bool: + total = 0 + for i, ch in enumerate(reversed(digits)): + d = ord(ch) - 48 + if i % 2 == 1: + d *= 2 + if d > 9: + d -= 9 + total += d + return total % 10 == 0 + + +class ApeEvent(_Strict): + ape_version: Literal["0.1"] + id: EventId + type: Literal[ + "payment.requested", + "payment.authorized", + "payment.settled", + "payment.failed", + "payment.refunded", + "payment.disputed", + ] + occurred_at: DateTime + environment: Literal["test", "live"] + source: Source + agent: Agent + owner: Owner + instruction: Instruction | None = None + payment: Payment + payee: Payee + instrument: Instrument | None = None + evidence: Annotated[list[Evidence], Field(max_length=16)] | None = None + related: Related | None = None + labels: Labels | None = None + + @model_validator(mode="after") + def _no_card_numbers(self) -> "ApeEvent": + texts = [ + self.payment.description, + self.payment.category, + self.payment.failure_reason, + self.instruction.summary if self.instruction else None, + self.agent.name, + self.owner.display_name, + self.payee.name, + *((self.labels or {}).values()), + ] + if any(t is not None and contains_pan(t) for t in texts): + raise ValueError("pan_detected") + return self + + +_EVENT = TypeAdapter(ApeEvent) + + +def validate_event(event: Mapping[str, Any]) -> list[str]: + """Problems with the event, as `path: message` strings. Empty when the event is valid.""" + try: + _EVENT.validate_python(dict(event)) + return [] + except ValidationError as exc: + return [".".join(str(p) for p in err["loc"]) + ": " + err["msg"] for err in exc.errors()] diff --git a/pyproject.toml b/pyproject.toml index 8054c7c..b01ae8c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "halvorai-ingest" -version = "0.1.0" +version = "0.2.0" description = "Signed, attributed payment events for AI agents. Never in the payment path." readme = "README.md" requires-python = ">=3.10" diff --git a/tests/test_client.py b/tests/test_client.py index d2dc835..3427758 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from typing import Any from pathlib import Path from halvorai_ingest import HalvorClient, agent_context, verify_event @@ -10,7 +11,7 @@ def test_emit_flush_sink_signs() -> None: data = json.loads(VECTOR.read_text(encoding="utf-8")) - envelopes: list[dict] = [] + envelopes: list[dict[str, Any]] = [] client = HalvorClient( source_key=data["private_jwk"], environment="test", @@ -56,5 +57,5 @@ def test_emit_flush_sink_signs() -> None: def test_emit_never_throws() -> None: client = HalvorClient(source_key=None) - assert client.emit(None) # type: ignore[arg-type] - assert client.emit("bad") # type: ignore[arg-type] + assert client.emit(None) + assert client.emit("bad") diff --git a/tests/test_quickstart.py b/tests/test_quickstart.py index d4413d2..71d2356 100644 --- a/tests/test_quickstart.py +++ b/tests/test_quickstart.py @@ -35,7 +35,7 @@ def test_offline_quickstart_round_trip() -> None: assert len(envelopes) == 1 event = envelopes[0]["event"] - assert event["source"] == {"id": "src_local", "sdk": "halvorai-ingest@0.1.0", "adapter": "custom"} + assert event["source"] == {"id": "src_local", "sdk": "halvorai-ingest@0.2.0", "adapter": "custom"} assert event["agent"]["id"] == "agt_travel_12" assert "text" not in event["instruction"] assert event["instruction"]["hash"].startswith("sha256:") diff --git a/tests/test_validate_and_delivery.py b/tests/test_validate_and_delivery.py new file mode 100644 index 0000000..6c4228c --- /dev/null +++ b/tests/test_validate_and_delivery.py @@ -0,0 +1,171 @@ +"""Validation before signing, environment-aware endpoints and per-event delivery results.""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from halvorai_ingest import HalvorClient, agent_context, generate_source_key +from halvorai_ingest.client import events_url +from halvorai_ingest.validate import validate_event + +PAYMENT = { + "payment_id": "pi_3QxAbCdEfGhIjKlM", + "rail": "stripe", + "amount": {"value": "41200", "unit": "minor", "currency": "USD"}, + "status": "settled", + "rail_ref": {"payment_intent": "pi_3QxAbCdEfGhIjKlM"}, +} +EVENT = {"type": "payment.settled", "payment": PAYMENT, "payee": {"type": "merchant", "name": "Example Airways"}} +CONTEXT: dict[str, Any] = { + "agent": {"id": "agt_travel_12"}, + "owner": {"id": "own_u_4410", "type": "person"}, +} + + +def collect() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + return [], [] + + +def test_malformed_event_is_dropped_and_reported_not_signed() -> None: + sent, errors = collect() + client = HalvorClient( + source_key=generate_source_key(), sink=lambda e: sent.extend(e), on_error=errors.append + ) + client.emit({"type": "payment.settled", "payment": {"amount": "lots"}}) + client.flush() + assert sent == [] + assert client.stats().dropped_invalid == 1 + assert client.stats().accepted == 0 + assert errors[0]["code"] == "invalid_event" + assert errors[0]["details"] + + +def test_valid_event_passes_and_is_signed() -> None: + sent, errors = collect() + client = HalvorClient( + source_key=generate_source_key(), sink=lambda e: sent.extend(e), on_error=errors.append + ) + with agent_context(**CONTEXT): + client.emit(EVENT) + client.flush() + assert errors == [] + assert len(sent) == 1 + assert validate_event(sent[0]["event"]) == [] + + +@pytest.mark.parametrize( + ("mutate", "where"), + [ + (lambda e: e["payment"].update(status="failed"), "failure_reason"), + (lambda e: e["payment"].pop("rail_ref"), "rail_ref"), + (lambda e: e["payee"].pop("name"), "payee"), + (lambda e: e["payment"]["amount"].update(value="41.20"), "value"), + (lambda e: e.update(type="payment.approved"), "type"), + ], +) +def test_rejects_what_the_schema_rejects(mutate: Any, where: str) -> None: + sent, errors = collect() + client = HalvorClient( + source_key=generate_source_key(), sink=lambda e: sent.extend(e), on_error=errors.append + ) + event = json.loads(json.dumps(EVENT)) + mutate(event) + with agent_context(**CONTEXT): + client.emit(event) + client.flush() + assert sent == [] + assert where in " ".join(errors[0]["details"]) + + +def test_card_numbers_are_redacted_before_validation_like_the_typescript_sdk() -> None: + sent, errors = collect() + client = HalvorClient( + source_key=generate_source_key(), sink=lambda e: sent.extend(e), on_error=errors.append + ) + event = json.loads(json.dumps(EVENT)) + event["payee"]["name"] = "Airways card 4242 4242 4242 4242" + with agent_context(**CONTEXT): + client.emit(event) + client.flush() + assert errors == [] + assert "4242 4242 4242 4242" not in json.dumps(sent) + + +def test_test_keys_go_to_the_sandbox_and_live_keys_to_production(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("HALVOR_ENDPOINT", raising=False) + test = HalvorClient(api_key="hk_test_" + "a" * 32, source_key=generate_source_key()) + live = HalvorClient(api_key="hk_live_" + "a" * 32, source_key=generate_source_key()) + assert test.endpoint == "https://api.sandbox.halvorai.com" + assert live.endpoint == "https://api.halvorai.com" + assert live.environment == "live" + assert HalvorClient(environment="test").endpoint == "https://api.sandbox.halvorai.com" + + +def test_endpoint_is_a_base_url_like_the_typescript_sdk() -> None: + assert events_url("http://127.0.0.1:8787") == "http://127.0.0.1:8787/v1/events" + assert events_url("http://127.0.0.1:8787/") == "http://127.0.0.1:8787/v1/events" + assert events_url("http://127.0.0.1:8787/v1/events") == "http://127.0.0.1:8787/v1/events" + + +def test_reads_keys_and_endpoint_from_the_environment(monkeypatch: pytest.MonkeyPatch) -> None: + key = generate_source_key(kid="src_env") + monkeypatch.setenv("HALVOR_API_KEY", "hk_test_" + "b" * 32) + monkeypatch.setenv("HALVOR_SOURCE_KEY", json.dumps(key)) + monkeypatch.setenv("HALVOR_ENDPOINT", "http://127.0.0.1:8787") + client = HalvorClient() + assert client.api_key == "hk_test_" + "b" * 32 + assert client.endpoint == "http://127.0.0.1:8787" + assert client._source_id == "src_env" + + +def test_counts_each_event_by_the_status_the_api_returned() -> None: + seen: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["url"] = str(request.url) + ids = [e["event"]["id"] for e in json.loads(request.content)["events"]] + results = [ + {"id": ids[0], "status": "accepted", "chain_seq": 1}, + {"id": ids[1], "status": "duplicate"}, + {"id": ids[2], "status": "rejected", "error": {"code": "invalid_signature"}}, + ] + return httpx.Response(202, json={"results": results}) + + errors: list[dict[str, Any]] = [] + client = HalvorClient( + api_key="hk_test_" + "c" * 32, + source_key=generate_source_key(), + endpoint="http://127.0.0.1:8787", + transport=httpx.MockTransport(handler), + on_error=errors.append, + ) + with agent_context(**CONTEXT): + for _ in range(3): + client.emit(EVENT) + client.flush() + stats = client.stats() + assert seen["url"] == "http://127.0.0.1:8787/v1/events" + assert (stats.accepted, stats.duplicates, stats.rejected) == (1, 1, 1) + assert errors == [ + { + "code": "rejected", + "message": "server rejected event", + "eventId": errors[0]["eventId"], + "serverCode": "invalid_signature", + } + ] + + +def test_no_api_key_counts_the_events_it_could_not_send(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("HALVOR_API_KEY", raising=False) + errors: list[dict[str, Any]] = [] + client = HalvorClient(source_key=generate_source_key(), on_error=errors.append) + with agent_context(**CONTEXT): + client.emit(EVENT) + client.flush() + assert client.stats().dropped_disabled == 1 + assert errors[0]["code"] == "config" diff --git a/uv.lock b/uv.lock index c158a4b..a538ee9 100644 --- a/uv.lock +++ b/uv.lock @@ -323,7 +323,7 @@ wheels = [ [[package]] name = "halvorai-ingest" -version = "0.1.0" +version = "0.2.0" source = { editable = "." } dependencies = [ { name = "cryptography" },