Metadata-Version: 2.4
Name: sbn-sdk
Version: 0.13.3
Summary: Python SDK for the SmartBlocks Network — Atlas frontier provisioning, GEC compute, SnapChore integrity, governance, and more.
License: MIT
Keywords: smartblocks,sbn,snapchore,gec,attestation,integrity,governance,workflow,lattice
Author: SmartBlocks Team
Author-email: devrel@smartblocks.network
Requires-Python: >=3.10,<4.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Provides-Extra: work-queue
Requires-Dist: PyJWT (>=2.8)
Requires-Dist: cryptography (>=41.0)
Requires-Dist: httpx (>=0.27.0,<0.28.0)
Requires-Dist: rich (>=13.0)
Requires-Dist: sqlalchemy (>=2.0) ; extra == "work-queue"
Requires-Dist: typer (>=0.9)
Project-URL: Documentation, https://smartblocks.network/docs/sdk
Project-URL: Homepage, https://smartblocks.network
Project-URL: Repository, https://github.com/smartblocks-network/sbn-sdk
Description-Content-Type: text/markdown

# SmartBlocks Network Python SDK

Canonical Python client for the SBN infrastructure. Covers the full network
surface — gateway, SnapChore, console, and control plane.

For the internal `pgdag-py` execution wrapper, the package also exposes a
dedicated HTTP client:

```python
from sbn import PGDagPyClient
```

The same package also ships the shared `pgdag` execution kernel used by Tower
products and workflow runtimes:

```python
from pgdag import PGDAGRunner, DAGTemplate, DAGNode
```

Choose the surface by workload shape:

- telemetry or pulse streams -> slot event wrapper
- bounded exact-state work -> direct SmartBlock submission
- proof-gated workflows -> `pgdag` or `pgdag-py`
- SnapChore-only proof capture -> SnapChore client

Release/publish hygiene for this package is tracked in:

- [docs/ops/sdk_publish_checklist.md](/C:/Users/mycal/tower_projects/smartblocks_v2/docs/ops/sdk_publish_checklist.md)

## Which wrapper should I use?

For most client integrations, start with one of these two wrappers:

| If your frontier feels like... | Use | Why |
|---|---|---|
| a live stream of events over time | `client.open_slot(...)` | best for generation windows, BitBlock emission, progressive `seal()`, and final `close()` |
| a native object or branch that evolves | `client.open_surface(...)` | best for origin/branch/successor flows and consistent surface metadata |

Keep `client.snapchore` separate in your mental model:

- use SnapChore when you want local-first capture / verify / seal
- use `promote()` only when you explicitly want to hand that local proof into SBN later
- do not treat SnapChore as the third main wrapper for normal network lifecycle work

## Install

```bash
pip install sbn-sdk
# optional if you need pgdag work claiming helpers
pip install "sbn-sdk[work-queue]"
# or from source
cd sdk/python && pip install -e .
```

## Quick start

For most event-heavy product work, the safest default is the slot event
wrapper. If your product instead revolves around a canonical object surface,
skip to [Agnostic surface wrapper](#agnostic-surface-wrapper). The broad quick
start below shows the main client namespaces.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

# SnapChore — capture, verify, seal
captured = client.snapchore.capture({"event": "signup", "user": "u-42"})
client.snapchore.verify(captured["hash"], {"event": "signup", "user": "u-42"})
sealed = client.snapchore.seal({"event": "signup", "user": "u-42"}, domain="auth")

# Gateway — slots, receipts, attestations
slot = client.gateway.create_slot(worker_id="w-1", task_type="classify")
receipt = client.gateway.fetch_receipt(slot.receipt_id, read_version=2)
print(receipt.canonical_proof_hash)
print(receipt.canonical_view)
print(receipt.provenance_view)

# Console — API keys, usage, billing
keys = client.console.list_api_keys("proj-123")
usage = client.console.get_usage("proj-123")

# Control plane — rate plans, tenants, validators
plans = client.control_plane.list_rate_plans()
client.control_plane.create_tenant(
    name="Acme Corp",
    contact_email="ops@acme.co",
    aggregator_endpoint="https://agg.acme.co",
    rate_plan_id=plans[0].id,
)
```

## Hello World: SnapChore

This is the smallest honest SnapChore flow:

1. capture a canonical hash
2. verify it
3. seal a local SmartBlock
4. inspect the local block
5. optionally promote it into SBN later

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

payload = {"event": "hello_world", "message": "Hello, SnapChore"}

captured = client.snapchore.capture(payload)
print("hash:", captured["hash"])

verified = client.snapchore.verify(captured["hash"], payload)
print("valid:", verified["valid"])

sealed = client.snapchore.seal(payload, domain="hello.world")
print("block:", sealed["id"])
print("record:", sealed["smartblock_record_id"])

detail = client.snapchore.get_block(sealed["smartblock_record_id"])
print("lineage role:", detail.get("lineage_identity", {}).get("surface_role"))

# Optional later handoff into the SBN attestation lane
promotion = client.snapchore.promote(
    smartblock_record_id=sealed["smartblock_record_id"]
)
print("promotion status:", promotion["status"])
```

Important:

- `seal()` is local-first
- the block exists before any receipt exists
- `promote()` is the explicit SBN handoff boundary

### Artifact descriptor helper

When you want to attest an external file without bloating the payload model,
compute a small canonical artifact descriptor from the exact file bytes:

```python
from sbn import SbnClient, artifact_descriptor_from_file

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

artifact = artifact_descriptor_from_file(
    "docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    artifact_uri="repo://docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
)

captured = client.snapchore.capture({"artifact": artifact})
print(captured["hash"])
```

This gives you:

- exact file-byte identity via `artifact_sha256`
- a stable descriptor SnapChore hash
- a descriptor you can embed into a later SmartBlock payload before attestation

### Golden path: hash local file bytes -> seal local proof block

If what you really want is the normal builder flow:

1. hash the local file bytes
2. wrap them in a standard proof payload
3. seal a local SmartBlock with moment auth

use the SnapChore convenience wrapper directly:

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

sealed = client.snapchore.seal_artifact_file(
    "docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    artifact_uri="repo://docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    summary="exact closure support map ready for review",
    subject_ref="exact-closure-milestone",
    claim={"milestone_ref": "patent-support-map-ready"},
)

print(sealed["snapchore_hash"])
print(sealed["smartblock_record_id"])
```

This is the intended SnapChore-first golden path:

- file bytes are hashed first
- the sealed block carries the artifact descriptor
- moment auth is attached by the SnapChore seal path
- promotion into SBN still stays explicit and separate

You can also inspect the result in one stable local-proof shape before any
promotion:

```python
status = client.snapchore.local_proof_status(sealed)
print(status["status"])                # sealed_local
print(status["snapchore_hash"])
print(status["moment_auth"]["present"])
```

And when you do promote later, normalize the handoff the same way:

```python
promotion = client.snapchore.promote(
    smartblock_record_id=sealed["smartblock_record_id"],
    frontier_id="artifact.review",
)
handoff = client.snapchore.promotion_status(
    promotion,
    local_status=status,
)
print(handoff["status"])  # submitted | queued_for_attestation | attested
```

If the attested receipt should land in a specific SBN project, target it
explicitly at promotion time:

```python
promotion = client.snapchore.promote_to_project(
    smartblock_record_id=sealed["smartblock_record_id"],
    target_project_id="tenant-123",
    frontier_id="artifact.review",
)
print(promotion["target_project_id"])
```

And if you need to create the target SBN project first:

```python
result = client.snapchore.create_project_and_promote(
    project_name="Artifact Lane",
    contact_email="ops@example.com",
    aggregator_endpoint="https://agg.example.com",
    rate_plan_id="plan-sandbox",
    smartblock_record_id=sealed["smartblock_record_id"],
    frontier_id="artifact.review",
)
print(result["target_project_id"])
print(result["promotion"]["promotion_status"])
```

If you want one copy-paste task-oriented example for this same lane, run:

```bash
py -3 scripts/examples/snapchore_artifact_task_flow.py docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md --api-key <sbn_live_key> --task-ref exact-closure-milestone --target-project-id <target_sbn_project_uuid>
```

### Canonical task -> artifact -> audit flow

If you want one infrastructure-level proof lane that builders can reuse across
products, use this shape:

1. `TaskBlock` records `task_started`
2. SnapChore seals the completed artifact
3. explicit promotion yields the attested receipt anchor
4. `AuditBlock` records `reconciled`

Minimal payload convention:

- `TaskBlock`
  - `task_ref`
  - `task_state`
  - recommended: `workflow_ref`, `operator_ref`, `subject_ref`
- SnapChore proof claim
  - `task_ref`
  - `task_state = artifact_completed`
  - `task_block_id`
- `AuditBlock`
  - `subject_ref`
  - `evidence_ref`
  - `reviewer_class`
  - recommended: `review_state`, `reconciliation_status`, `task_ref`

Recommended shared `task_state` ladder:

- `task_declared`
- `task_started`
- `task_in_progress`
- `artifact_expected`
- `artifact_completed`
- `task_closed`

Keep `reconciled` on the `AuditBlock`, not the `TaskBlock`.

Default allowed transitions:

- `task_declared -> task_started`
- `task_started -> task_in_progress`
- `task_started -> artifact_expected`
- `task_in_progress -> artifact_expected`
- `task_in_progress -> artifact_completed`
- `artifact_expected -> artifact_completed`
- `artifact_completed -> task_closed`

Reasonable shortcuts:

- `task_declared -> artifact_expected`
- `task_started -> artifact_completed`

Runnable example:

```bash
py -3 scripts/examples/task_artifact_audit_flow.py docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md --api-key <sbn_live_key> --task-ref task-42 --target-project-id <target_sbn_project_uuid>
```

It writes the latest builder report to:

- `artifacts/examples/task_artifact_audit_latest.json`

That example:

- creates a direct `TaskBlock`
- seals the artifact through SnapChore
- promotes into SBN
- reads back the receipt anchor
- emits an `AuditBlock` tied to that receipt
- returns a normalized `receipt_anchor` payload for downstream app use
- if the public receipt catalogue has to degrade malformed rows in place, ops
  can inspect `/health/degraded` for `aggregator.receipt_summary`
- polls queue state long enough to emit a realistic `proof_ladder.final_status`
  and queue summary for builder reports

The shared contract for this lane lives in:

- `docs/SBN_TASK_ARTIFACT_AUDIT_FLOW.md`

Builder helper imports:

```python
from sbn import (
    validate_finance_block_metadata,
    validate_task_block_metadata,
    validate_task_state_transition,
    validate_audit_block_metadata,
    validate_audit_review_transition,
)

validate_task_block_metadata(
    {"task_ref": "task-42", "task_state": "artifact_completed"},
    previous_state="task_started",
)
validate_audit_block_metadata(
    {
        "subject_ref": "task-42",
        "evidence_ref": "receipt:abc123",
        "reviewer_class": "ops",
        "review_state": "reconciled",
    },
    previous_state="review_in_progress",
)
validate_finance_block_metadata(
    {
        "lifecycle_state": "settlement_observed",
        "instrument_ref": "payable-100",
        "position_ref": "position-100",
        "counterparty_ref": "merchant-1",
        "financial_interop": {
            "standard": "iso20022",
            "business_domain": "payments",
        },
    },
    previous_state="obligation_attached",
)
```

If you want the reusable cross-project smoke for operators, run:

```bash
py -3 scripts/smoke/snapchore_target_project_smoke.py --api-key <sbn_live_key> --target-project-id <target_sbn_project_uuid> --service-api-key <service_client_secret>
```

Raw runtime states collapse into the normalized ladder like this:

- `sealed_only` / `local_only` -> `sealed_local`
- `promotion_requested` / `submitted_to_sbn` -> `submitted`
- `handoff_prepared` / `queued_for_attestation` -> `queued_for_attestation`
- any result carrying `receipt_hash` or `receipt_id` -> `attested`

For receipt linkage, prefer the shared resolver:

```python
receipt_anchor = client.snapchore.resolve_receipt_anchor(
    handoff=handoff,
    promotion_result=promotion,
    block_detail=block_detail,
    receipt_lookup=receipt_lookup,
    target_project_id="your-project-id",
)
print(receipt_anchor["evidence_ref"])
```

If you want a tiny runnable ladder example instead of copying snippets, use:

```bash
py -3 scripts/examples/snapchore_proof_ladder.py docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md --api-key <sbn_live_key> --promote --frontier-id artifact.review
```

It writes:

- `artifacts/examples/snapchore_proof_ladder_latest.json`

If you want the reusable post-deploy smoke for this same handoff lane, run:

```bash
py -3 scripts/smoke/snapchore_attestation_smoke.py --api-key <sbn_live_key>
```

It writes:

- `artifacts/smoke/snapchore_attestation_latest.json`

and proves the normalized ladder all the way through:

- `sealed_local`
- `submitted` / `queued_for_attestation`
- `attested`

### End-to-end artifact attestation example

```python
from sbn import SbnClient, artifact_descriptor_from_file

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

artifact = artifact_descriptor_from_file(
    "docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
    artifact_uri="repo://docs/EXACT_CLOSURE_PATENT_SUPPORTING_ARTIFACT_MAP.md",
)

block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": "exact-closure-milestone",
            "evidence_ref": "artifact:exact-closure-map",
            "reviewer_class": "ops",
        },
        "artifact": artifact,
        "claim": {
            "milestone_ref": "patent-support-map-ready",
            "summary": "supporting artifact map assembled for counsel handoff",
        },
    },
)

detail = client.blocks.get(block["data"]["id"])
receipt_hash = detail.get("hash") or detail.get("canonical_proof_hash")

if receipt_hash:
    receipt = client.receipts.get_by_hash(
        receipt_hash,
        project_id="your-project-id",
    )
    print(receipt["items"][0]["native_receipt"]["subject_ref"])
```

## Hello World: SBN

This is the smallest native SBN flow using the agnostic surface wrapper:

1. open an origin surface
2. emit one event
3. seal canonical local state
4. close the stream
5. inspect the resulting proof/receipt surfaces

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

with client.open_surface(
    worker_id="hello-agent",
    surface_family="hello.world",
    surface_id="hello-world-001",
    metadata={"frontier_hint": "hello.world"},
) as surface:
    surface.pulse(
        "hello_event",
        score=1.0,
        actions=[{"action": "hello_event", "score": 1.0}],
        message="Hello, SBN",
    )

    seal = surface.seal()
    print("anchor hash:", seal.anchor_hash)

close_receipt = surface.receipt
print("canonical proof hash:", close_receipt.canonical_proof_hash)
print("canonical view:", close_receipt.canonical_view)
print("provenance view:", close_receipt.provenance_view)
```

Important:

- the native scalar law stays `r = Y / X`, `g = r / c_max`
- frontier contracts declare what `Y`, `X`, and `c_max` mean for that ecology
- `X` is customizable
- timed generation windows are a strong live-telemetry default, not a universal
  framework rule

Use this path when you want:

- canonical object surfaces
- branch and successor behavior later
- proof-bearing close/seal lifecycle
- receipts as first-class network outputs

## `pgdag-py` wrapper

Use `PGDagPyClient` when you want to talk directly to the `services/pgdag_py`
HTTP wrapper instead of embedding `pgdag` in-process.

```python
from sbn import PGDagPyClient

client = PGDagPyClient(base_url="http://localhost:8083")
health = client.health_live()

result = client.run(
    template={
        "name": "kinetic_integrity",
        "version": "1.0",
        "domain": "kinetic.session.integrity.v1",
        "nodes": [
            {
                "id": "integrity_eval",
                "label": "Integrity eval",
                "metadata": {"builtin": "pass_inputs"},
            }
        ],
        "edges": [],
    },
    inputs={"session_id": "sess-42"},
    actor="tower-pgdag-py",
)

print(health["status"])
print(result["slot_id"], result["slot_receipt_id"])
```

When the `pgdag-py` service itself has `SBN_BASE_URL` plus valid SBN
credentials, that run publishes the standard open -> BitBlock -> seal -> close
lifecycle through SBN.

## Event streaming wrapper

For telemetry-style frontiers, use the slot session wrapper instead of
manually wiring slot open, BitBlock submission, and close on every project.
This is the preferred developer surface when:

- your app emits event pulses over time
- generation windows matter
- you want progressive `seal()` boundaries
- you want to avoid repeated auth/lifecycle mistakes across integrations

Mental model:

- one open slot = one live operational window
- BitBlocks capture what happened during that window
- `seal()` gives you progressive proof anchors while work continues
- `close()` gives you the final slot-summary receipt
- explicit extra attestation is optional and separate

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")
client.set_tenant("tower-agents")

with client.open_slot(
    worker_id="tower-agents",
    task_type="grid.atomic_rentals",
    domain="grid.atomic_rentals.streaming",
    generation_interval=1800,  # 30 minutes
    metadata={"frontier_hint": "grid.atomic_rentals.streaming"},
) as stream:
    stream.pulse(
        "delivery_picked_up",
        score=0.9,
        actions=[{"action": "delivery_picked_up", "score": 0.9}],
        interactions=[{"action": "route_quality", "score": 0.94}],
        events={"friendly": 1},
        booking_id="bk_123",
        status="picked_up",
        metadata={"interaction_id": "rental-123"},
        operation_id="tower-outbox-123",
        idempotency_key="sha256:stable-provider-command",
    )
    seal = stream.seal()
    print(seal.anchor_hash, seal.anchor_type)
    print(seal.proof_chain)
    print(seal.operation_refs)

    stream.pulse(
        "delivery_complete",
        score=1.0,
        actions=[{"action": "delivery_complete", "score": 1.0}],
        events={"friendly": 1},
        booking_id="bk_123",
        status="complete",
        metadata={"interaction_id": "rental-123"},
    )

close_receipt = stream.receipt
print(close_receipt.operation_id)
print(close_receipt.idempotency_key)
print(close_receipt.operation_refs)
print(close_receipt.canonical_proof_hash)
print(close_receipt.native_proof_chain)
print(close_receipt.lifecycle_contract)
print(close_receipt.canonical_view)
print(close_receipt.provenance_view)
```

Operation identity is evidence correlation only. It does not prove that SBN
executed an external side effect or that the external provider enforces the
idempotency key. Use provider receipt retrieval and verification before
reconciling an uncertain action as successful.

### Finance example

Use a `FinanceBlock` when the app wants to record a financial lifecycle state,
not when it wants SBN to settle or custody value.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

finance_block = client.blocks.create(
    domain="finance.core",
    payload={
        "type": "FinanceBlock",
        "domain": "finance.core",
        "metadata": {
            "lifecycle_state": "settlement_observed",
            "instrument_ref": "payable-100",
            "external_settlement_ref": "sonicpay-8841",
            "observed_at": "2026-06-07T10:30:00Z",
        },
    },
)

detail = client.blocks.get(finance_block["data"]["id"])
print(detail["block_type"])  # finance
print(detail["block_type_profile"]["scalar_emphasis"])  # muted
print(detail["scalar_gec"])  # None on public read surfaces
```

### Audit example

Use an `AuditBlock` when Ops, NUMA, or an SBN cadence process wants to record
review or reconciliation state.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

audit_block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": "finance-surface-demo-1",
            "evidence_ref": "receipt:abc123",
            "reviewer_class": "ops",
            "reconciliation_status": "reconciled",
            "reviewed_at": "2026-06-07T10:45:00Z",
        },
    },
)

detail = client.blocks.get(audit_block["data"]["id"])
print(detail["block_type"])  # audit
print(detail["block_type_profile"]["authority_posture"])
print(detail["block_type_profile"]["minimal_fields"])
```

### App-oriented slot pattern

One common app pattern is:

1. submit several `labor` blocks into an open slot over a work window
2. submit one `finance` block describing the resulting financial state transition
3. close the slot

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

with client.open_slot(
    worker_id="delivery-agent",
    task_type="delivery.window",
    domain="gig.delivery",
    generation_interval=1800,
) as stream:
    for delivery_id in ("d1", "d2", "d3", "d4"):
        client.blocks.submit(
            slot_id=stream.slot_id,
            domain="gig.delivery",
            payload={
                "type": "LaborBlock",
                "domain": "gig.delivery",
                "metadata": {
                    "delivery_id": delivery_id,
                    "completion_state": "completed",
                    "window_ref": "delivery-window-2026-06-07",
                },
            },
        )

    client.blocks.submit(
        slot_id=stream.slot_id,
        domain="finance.core",
        payload={
            "type": "FinanceBlock",
            "domain": "finance.core",
            "metadata": {
                "lifecycle_state": "settlement_observed",
                "instrument_ref": "delivery-window-2026-06-07",
                "external_settlement_ref": "sonicpay-cashout-1",
                "observed_amount": 100,
            },
        },
    )
close_receipt = stream.receipt
print(close_receipt.canonical_proof_hash)
```

This keeps the contract clean:

- the app decides when the labor window is complete
- the app decides when the financial state transition was observed
- SBN attests the lifecycle and groups the resulting blocks under one slot close

### Paired finance -> audit follow-up

Use this when a finance state later needs authoritative review.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

finance_block = client.blocks.create(
    domain="finance.core",
    payload={
        "type": "FinanceBlock",
        "domain": "finance.core",
        "metadata": {
            "lifecycle_state": "settlement_observed",
            "instrument_ref": "delivery-window-2026-06-07",
            "external_settlement_ref": "sonicpay-cashout-1",
        },
    },
)

audit_block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": finance_block["data"]["id"],
            "evidence_ref": "receipt:abc123",
            "reviewer_class": "ops",
            "reconciliation_status": "reconciled",
        },
    },
)

print(audit_block["data"]["id"])
```

### Receipt lookup after finance -> audit

Use this when you want to inspect the read surface produced by the finance and
audit flow.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

lookup = client.receipts.get_by_hash(
    "receipt-hash",
    project_id="project-id",
)
item = (lookup.get("items") or [{}])[0]

print(item.get("block_type_profile", {}).get("block_type"))
print(item.get("trust_summary", {}).get("trust_class"))
print(item.get("cdna_state", {}).get("closure_regime"))
print(item.get("mutation_summary", {}).get("decision_divergence"))
```

### First law example from the same base

`law` is still maturing, but this is the intended follow-up shape when a
finance or audit state needs an explicit obligation or constraint record.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

law_block = client.blocks.create(
    domain="law.core",
    payload={
        "type": "LawBlock",
        "domain": "law.core",
        "metadata": {
            "law_state": "bound",
            "subject_ref": "finance-or-audit-subject-id",
            "constraint_ref": "derivative-exercise-window-1",
            "issuer_class": "compliance_service",
            "obligation_code": "exercise_window_active",
            "binding_scope": "derivative_position",
        },
    },
)

print(law_block["data"]["id"])
```

### Law lifecycle proposal

The smallest honest lifecycle for `law` right now is:

```text
declared -> bound -> active -> satisfied | expired
```

Use it as:

- `declared` for a newly stated rule or obligation
- `bound` when attached to a subject or position
- `active` while in force
- `satisfied` when fulfilled
- `expired` when no longer in force

### Narrative pattern: labor -> finance -> audit -> law

One clean app-level narrative is:

1. submit `labor` blocks for the productive window
2. submit a `FinanceBlock` for the observed value-state transition
3. submit an `AuditBlock` for reconciliation
4. submit a `LawBlock` for the resulting obligation or constraint

### Derivative-specific example: finance -> audit -> law

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

finance_block = client.blocks.create(
    domain="finance.core",
    payload={
        "type": "FinanceBlock",
        "domain": "finance.core",
        "metadata": {
            "lifecycle_state": "obligation_attached",
            "instrument_ref": "eth-call-option-2026-q3",
            "position_ref": "position-eth-call-1",
            "counterparty_ref": "desk-a",
        },
    },
)

audit_block = client.blocks.create(
    domain="audit.core",
    payload={
        "type": "AuditBlock",
        "domain": "audit.core",
        "metadata": {
            "review_state": "reconciled",
            "subject_ref": finance_block["data"]["id"],
            "evidence_ref": "receipt:derivative-book-1",
            "reviewer_class": "ops",
            "reconciliation_status": "reconciled",
        },
    },
)

law_block = client.blocks.create(
    domain="law.core",
    payload={
        "type": "LawBlock",
        "domain": "law.core",
        "metadata": {
            "law_state": "active",
            "subject_ref": finance_block["data"]["id"],
            "constraint_ref": "exercise-window-2026-q3",
            "issuer_class": "compliance_service",
            "obligation_code": "cash_settlement_if_exercised",
            "binding_scope": "derivative_position",
            "effective_at": "2026-07-01T00:00:00Z",
            "expires_at": "2026-09-30T23:59:59Z",
        },
    },
)

print(audit_block["data"]["id"], law_block["data"]["id"])
```

### Shared derivative payload convention

Prefer this shared reference vocabulary across the derivative lifecycle:

- `instrument_ref`
- `position_ref`
- `counterparty_ref`
- `constraint_ref`
- `evidence_ref`

Recommended by type:

1. `FinanceBlock`
   - `instrument_ref`
   - `position_ref`
   - `counterparty_ref`
2. `AuditBlock`
   - `subject_ref`
   - `evidence_ref`
3. `LawBlock`
   - `subject_ref`
   - `constraint_ref`
   - `effective_at`
   - `expires_at`

### Receipt surfaces

Gateway receipts now follow the same three-surface model used across block and
receipt reads:

- `canonical_view` - default proof truth
- `provenance_view` - source ids and compatibility lineage
- `forensic_view` - raw receipt body and transport-era detail when explicitly
  needed

As a rule, SDK examples should read canonical view first and inspect forensic
view only when debugging or auditing.

### Read versions

Gateway read methods now support an explicit compatibility seam:

- `read_version=2` (default) returns the canonical-first contract
- `read_version=1` restores older top-level compatibility aliases for legacy
  consumers

Treat `read_version=2` as the stable/public contract for new integrations.
Use `read_version=1` only when migrating older consumers or when internal
ops/debug tooling still needs the legacy top-level aliases.

Examples:

```python
block = client.gateway.fetch_block(block_id, read_version=2)
receipt = client.gateway.fetch_receipt(receipt_id, read_version=2)
legacy_receipt = client.gateway.fetch_receipt(receipt_id, read_version=1)
```

### Stream profile examples

Use `stream_profile` when you want the slot wrapper to stamp a deterministic
frontier contract and canonical block-type posture without hand-building the
entire contract object first.

#### `labor_v1`

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

with client.open_slot(
    worker_id="grid-worker-1",
    task_type="gig.delivery.window",
    domain="gig.delivery",
    generation_interval=30,
    stream_profile="labor_v1",
    metadata={
        "frontier_hint": "grid.atomic_rentals",
        "proof_intent": "delivery_window",
    },
) as stream:
    stream.yield_pulse(
        "delivery_complete",
        score=1.0,
        actions=[{"action": "delivery_complete", "score": 1.0}],
        events={"friendly": 1},
        state_hint="delivery_complete",
        delivery_state="completed",
    )
    stream.burden_pulse(
        "late_delivery",
        burden_hint=0.35,
        score=0.2,
        actions=[{"action": "late_delivery", "score": 0.2}],
        events={"hostile": 1},
        severity="warning",
        state_hint="late_delivery",
        delivery_state="late",
    )
    seal = stream.seal()

final = stream.finalize(
    request_attestation=True,
    snap_hash=str(seal.anchor_hash or ""),
    frontier_id="grid.atomic_rentals",
    metadata={"type": "LaborBlock", "domain": "gig.delivery"},
)
```

Expected posture:

- composed block resolves as `labor`
- public block read exposes `scalar_gec`
- receipt and block both surface `bb_contribution_summary`

#### `ops_audit_v1`

```python
with client.open_slot(
    worker_id="ops-review-1",
    task_type="ops.audit.review.window",
    domain="audit.core",
    generation_interval=30,
    stream_profile="ops_audit_v1",
    metadata={
        "frontier_hint": "ops.audit.review",
        "proof_intent": "scheduled_review",
    },
) as stream:
    stream.yield_pulse(
        "cadence_snapshot",
        score=1.0,
        actions=[{"action": "cadence_snapshot", "score": 1.0}],
        events={"neutral": 1},
        state_hint="cadence_snapshot",
        review_state="cadence_snapshot",
        subject_ref="ops.audit.review",
        evidence_ref="frontier:ops.audit.review",
        reviewer_class="sbn_timer",
    )
    stream.burden_pulse(
        "review_in_progress",
        burden_hint=0.5,
        score=0.2,
        actions=[{"action": "review_in_progress", "score": 0.2}],
        events={"hostile": 1},
        severity="warning",
        state_hint="review_in_progress",
        review_state="review_in_progress",
        subject_ref="ops.audit.review",
        evidence_ref="frontier:ops.audit.review:delta",
        reviewer_class="ops",
    )
```

Expected posture:

- composed block resolves as `audit`
- `block_type_profile.grouping_pattern = "scheduled_or_exception_review"`
- `block_type_profile.allowed_emitters` includes `ops`, `numa`, and `sbn_timer`

#### `finance_state_v1`

```python
with client.open_slot(
    worker_id="finance-state-1",
    task_type="finance.state.window",
    domain="finance.core",
    generation_interval=30,
    stream_profile="finance_state_v1",
    metadata={
        "frontier_hint": "finance.state.window",
        "proof_intent": "financial_lifecycle_state",
    },
) as stream:
    stream.yield_pulse(
        "allocation",
        score=1.0,
        actions=[{"action": "allocation", "score": 1.0}],
        events={"friendly": 1},
        state_hint="allocation",
        lifecycle_state="allocation",
        instrument_ref="position-1:instrument",
        position_ref="position-1",
        counterparty_ref="venue.alpha",
    )
    stream.yield_pulse(
        "obligation_attached",
        score=0.9,
        actions=[{"action": "obligation_attached", "score": 0.9}],
        events={"friendly": 1},
        state_hint="obligation_attached",
        lifecycle_state="obligation_attached",
        instrument_ref="position-1:instrument",
        position_ref="position-1",
        counterparty_ref="venue.alpha",
        obligation_ref="position-1:obligation",
    )
    stream.burden_pulse(
        "settlement_observed",
        burden_hint=0.25,
        score=0.4,
        actions=[{"action": "settlement_observed", "score": 0.4}],
        events={"hostile": 1},
        severity="warning",
        state_hint="settlement_observed",
        lifecycle_state="settlement_observed",
        instrument_ref="position-1:instrument",
        position_ref="position-1",
        counterparty_ref="venue.alpha",
        settlement_ref="position-1:settlement",
        external_system_ref="sonicpay",
    )
```

Expected posture:

- composed block resolves as `finance`
- public block read mutes `scalar_gec`
- `block_type_profile.settlement_posture = "observe_reference_reconcile"`
- `block_type_profile.external_system_role = "adjacent_non_settlement"`

### Metric taxonomy

SBN now treats block-side quantitative state in four classes:

- local descriptive metrics
  - `yield`, `entropy`, `compression`, `trust`, `reflex`, local `metrics.gec`
- canonical reduction inputs
  - lifecycle-supporting evidence used to derive carrier state
- canonical lifecycle state
  - `carrier_state`, `carrier_level`, branch/surface/root state
- parent-facing widening context
  - `widening` and related sufficiency context

That means not every numeric field on a block should be read as canonical
lifecycle truth. Local metrics describe; carrier state carries; widening
qualifies upstream sufficiency.

### Shared GEC operation contract

When you want SBN to derive default `y` and `x` from the same operation
metadata everywhere, pass a shared contract into the slot wrapper.

If you want a more builder-friendly SmartBlock-native path, define the
frontier contract first and hand the result into either `open_slot(...)`
or `open_surface(...)`. Atlas can later catalogue the same contract, but
the helper itself is not Atlas-owned.

```python
from sbn import FrontierContractBuilder, SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")

contract = (
    FrontierContractBuilder(
        frontier="assistant",
        gec_domain="ai",
        app_name="assistant-demo",
    )
    .archetype("stream_efficiency")
    .native_defaults()
    .canonical_cost_profile("ai_assistant_v1")
    .actions("prompt_submitted", "tool_call_completed", "response_delivered", weights={
        "prompt_submitted": 0.25,
        "tool_call_completed": 0.6,
        "response_delivered": 1.0,
    })
    .events("assistant_turn", weights={
        "friendly": 1.0,
        "neutral": 0.0,
        "hostile": -1.0,
    })
    .strategy("weighted_mean")
    .yield_rule("completed_turns", unit="assistant_turn")
    .cost_rule(
        "blended",
        composition_rule="weighted_sum_v1",
        required_components=["compute"],
        optional_components=["time", "resource", "intervention", "reversal_risk"],
    )
    .cmax_rule("frontier_registry")
    .interpretation(
        entropy_mode="structural_entropy_v1",
        compression_mode="canonical_compression_v1",
        trust_mode="provenance_and_coherence",
        reflex_mode="trajectory_delta",
    )
    .build()
)

with client.open_surface(
    worker_id="assistant-demo",
    surface_family="ai.assistant",
    surface_id="conversation-001",
    gec_contract=contract,
) as surface:
    surface.event({"event_type": "prompt_submitted", "score": 0.25})
```

```python
from sbn import SbnClient, GecOperationContract

contract = GecOperationContract(
    app_name="assistant-demo",
    gec_domain="ai",
    frontier="assistant",
    frontier_archetype="stream_efficiency",
    phase="response_generation",
    x_mode="elapsed_time",
    action_weight_map={"response_delivered": 1.0},
    interaction_weight_map={"tool_call_completed": 0.6},
    event_polarity_weights={"friendly": 1.0, "neutral": 0.1, "hostile": -1.0},
    native_metric_profile="native_gec_v1",
    yield_mode="completed_turns",
    yield_unit="assistant_turn",
    cost_mode="blended",
    cost_composition_rule="weighted_sum_v1",
    cost_components={
        "time": 0.5,
        "compute": 1.0,
        "intervention": 0.75,
        "reversal_risk": 0.25,
    },
    cost_required_components=["compute"],
    cost_optional_components=["time", "resource", "intervention", "reversal_risk"],
    c_max_mode="frontier_registry",
    entropy_mode="structural_entropy_v1",
    compression_mode="canonical_compression_v1",
    trust_mode="provenance_and_coherence",
    reflex_mode="trajectory_delta",
    native_notes={"canonical_cost_profile": "ai_assistant_v1"},
)

with client.open_slot(
    worker_id="assistant-demo",
    task_type="ai.assistant.response_generation",
    domain="ai.assistant",
    generation_interval=60,
    gec_contract=contract,
) as stream:
    ...
```

When a live telemetry frontier uses `generation_interval` as its base `X`
coordinate, read that as a frontier declaration:

- `X = elapsed generation window`
- `Y = persisted useful output inside that window`

That keeps interval comparisons stable and easy to interpret. Other frontiers
may instead declare another lawful extensive `X` basis, such as labor, compute,
tokens, or capital.

You can also compose an observed burden packet against the contract directly:

```python
observed_cost = contract.compose_cost_observation(
    {
        "time": 2.0,
        "compute": 1.5,
        "intervention": 0.5,
    }
)

print(observed_cost["rule"])   # weighted_sum_v1
print(observed_cost["total"])  # 2.875
print(observed_cost["valid"])  # True
```

For a generic binary approval frontier, use the public-safe fixed-burden profile:

```python
approval_contract = client.define_frontier(
    frontier="approval",
    gec_domain="workflow",
    frontier_archetype="binary_phase",
    apply_native_defaults=True,
    canonical_cost_profile="workflow_approval_v1",
)
```

### Agnostic surface wrapper

When you want developers to think in terms of a canonical object surface
instead of raw slot metadata, use `open_surface(...)`.

This keeps the same transport under the hood, but it writes the native
surface/branch metadata consistently for you:

- origin surfaces omit `fractal_parent_block_id`
- branch surfaces set `fractal_parent_block_id`
- branch surfaces default to `fractal_auto_rollup=True`
- `surface_family` becomes the default `task_type` / `domain`

Mental model:

- slots stay the operational transport
- surfaces give that transport a native object identity
- origin/branch/successor structure becomes explicit
- `seal()` and `close()` still preserve the same proof-bearing lifecycle underneath

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

with client.open_surface(
    worker_id="grid-agent",
    surface_family="grid.atomic_rentals",
    surface_id="rental-001",
) as surface:
    surface.event({"event_type": "rental_created", "score": 1.0})
    root_seal = surface.seal()

with client.open_surface(
    worker_id="grid-agent",
    surface_family="grid.atomic_rentals",
    surface_id="rental-001.active",
    parent_block_id=root_seal.smartblock_id,
    branch_kind="active_rental",
) as branch:
    branch.event({"event_type": "vehicle_checkout", "score": 0.92, "weight": 1.0})
    branch_seal = branch.seal()
```

This is intentionally transport-agnostic:

- slots remain the operational wrapper
- BitBlocks remain the atomic evidence path
- successor surfaces remain the canonical continuation target
- public attestation still belongs on canonical surfaces, not raw branch snapshots

### cDNA mutation lineage

When your runtime needs explicit mutation or adaptation history, use the
public `sbn.cdna` helper path instead of reaching into the internal `pgdag`
package structure.

```python
from sbn.cdna import (
    EvolutionaryMutator,
    LineageManager,
    MutationStrategy,
    MutationTrigger,
)

trigger = MutationTrigger(
    strategy=MutationStrategy.ENTROPY_YIELD,
    yield_score=0.31,
    entropy_score=0.82,
    reasoning="low yield under chaotic conditions",
)

mutator = EvolutionaryMutator(
    yield_threshold=0.5,
    entropy_threshold=0.65,
)

# Your product supplies the concrete chain manager / proof writer.
lineage = LineageManager(
    chain_manager=my_chain_manager,
    writer=my_proof_writer,
)

record = lineage.seed("agent-42", "watcher:performance", spec_version="0.1.0")
proof = lineage.check_and_mutate(
    "agent-42",
    yield_score=trigger.yield_score,
    entropy_score=trigger.entropy_score,
    reasoning=trigger.reasoning,
)

if proof is not None:
    print(proof.before_cdna, "->", proof.after_cdna)
```

This maps directly to the existing core lifecycle:

- agent JWTs can carry a `cdna` claim
- SmartBlock creation injects `c_dna`
- cDNA may mutate once pre-sign during metrics computation
- lineage material is preserved in downstream BitBlock and fusion state

The same contract can be rendered for Atlas registration with:

```bash
sbn frontier workflow-spec \
  --workflow pushup_session \
  --app-name pt_web \
  --gec-domain kinetic \
  --frontier pushup \
  --constraint-tag amrap \
  --phase execution
```

### Close and finalize semantics

The wrapper now has three useful boundaries:

- `seal()` - compose the current pulse into a proof-bearing anchor while the slot stays open
- `close()` - close the slot and persist the final slot-summary receipt while preserving native lifecycle context
- `finalize()` - optionally request an additional explicit attestation with a
  caller-provided `snap_hash`

That means proof can become real during the stream instead of waiting until
the final close call.

The wrapper remains operational only. The native lifecycle truth still lives
underneath it, and the close/seal results preserve that structure:

- `close().native_proof_chain`
- `close().successor_surface_ids`
- `close().lifecycle_contract`
- `seal().proof_chain`

### Explicit attestation

When you need an extra billable / quorum attestation beyond progressive seals,
request it explicitly with the hash you want routed through the attestation
pipeline. The close-time receipt remains available even if the explicit
attestation path fails. In that case `final.attestation_error` explains the
failure while `final.close` still carries the successful slot close result.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")
client.set_tenant("tower-agents")

with client.open_slot(
    worker_id="tower-agents",
    task_type="grid.atomic_rentals",
    domain="grid.atomic_rentals.streaming",
) as stream:
    stream.pulse(
        "delivery_complete",
        score=1.0,
        actions=[{"action": "delivery_complete", "score": 1.0}],
        events={"friendly": 1},
        booking_id="bk_123",
    )
    step_anchor = stream.seal()

final = stream.finalize(
    request_attestation=True,
    snap_hash=step_anchor.anchor_hash,
    frontier_id="grid.atomic_rentals.streaming",
    metadata={
        "type": "LaborBlock",
        "domain": "grid",
        "event_type": "delivery_complete",
    },
)

print(final.close.receipt_id)
print(final.close.native_proof_chain)
print(final.close.lifecycle_contract)
print(final.attestation)
print(final.attestation_error)
```

If the explicit attestation is required to hard-fail the call, opt into strict
mode:

```python
final = stream.finalize(
    request_attestation=True,
    snap_hash=step_anchor.anchor_hash,
    frontier_id="grid.atomic_rentals.streaming",
    strict_attestation=True,
)
```

## Auth methods

```python
# API key (most common for external devs)
client.authenticate_api_key("sbn_live_...")

# Bearer token (console sessions, service-to-service)
client.authenticate_bearer("eyJ...")

# Ed25519 signing key (auto-refreshing JWTs for agents)
from sbn import SigningKey
key = SigningKey.from_pem("/path/to/key.pem", issuer="my-svc", audience="sbn")
client.authenticate_signing_key(key, scopes=["attest.write", "snapchore.seal"])
```

## Sub-clients

| Property | Domain | Key operations |
|----------|--------|----------------|
| `client.gateway` | Slots & receipts | `create_slot`, `close_slot`, `fetch_receipt`, `request_attestation` |
| `client.snapchore` | Hash capture | `capture`, `verify`, `seal`, `create_chain`, `append_to_chain` |
| `client.console` | Developer console | `list_api_keys`, `create_api_key`, `get_usage`, `get_billing_status` |
| `client.control_plane` | Multi-tenancy | `list_rate_plans`, `create_tenant`, `register_validator` |

## Legacy compatibility

The original `sbn_gateway.py` single-file SDK is preserved for backward
compatibility. New integrations should use `from sbn import SbnClient`.

## Atlas workflows

The Python SDK exposes Atlas app registration, key binding, and workflow
binding operations through `client.atlas`.

```python
from sbn import SbnClient

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

app = client.atlas.register_app(
    app_name="pt-kinetic",
    app_family="kinetic.sessions",
    declared_workflows=["pushup_session", "session_integrity"],
    workflow_specs=[
        {
            "workflow": "pushup_session",
            "frontier_id": "kinetic.pushup.amrap.v1",
            "pulse_type": "slot_stream",
            "event_types": ["rep", "set_complete"],
        },
        {
            "workflow": "session_integrity",
            "frontier_id": "kinetic.session.integrity.v1",
            "pulse_type": "pgdag",
            "event_types": ["seal_complete"],
        },
    ],
)

minted = client.atlas.mint_bound_key(app["app_id"], label="pt-kinetic network key")
client.atlas.upsert_binding(
    app["app_id"],
    workflow="pushup_session",
    frontier_id="kinetic.pushup.amrap.v1",
)
```

The CLI mirrors the same surface:

```bash
sbn atlas register \
  --name "pt-kinetic" \
  --family kinetic.sessions \
  --workflow pushup_session \
  --workflow session_integrity \
  --bind pushup_session=kinetic.pushup.amrap.v1 \
  --bind session_integrity=kinetic.session.integrity.v1 \
  --pulse pushup_session=slot_stream \
  --pulse session_integrity=pgdag

sbn atlas keys mint <app-id> --label "pt-kinetic network key"
sbn atlas keys bind <app-id> <api-key-id>
sbn atlas bindings set <app-id> --workflow pushup_session --frontier kinetic.pushup.amrap.v1
sbn atlas bindings history <app-id> --workflow pushup_session
sbn atlas bindings restore <app-id> --workflow pushup_session --index 2
```

### Atlas frontier cards from the CLI

Atlas also exposes the card-first workflow directly from the CLI, including
round-tripping card JSON for external authoring tools such as RI.

```bash
# inspect the private catalogue
sbn atlas cards list
sbn atlas cards show kinetic.pushup.amrap.v1

# register from flags
sbn atlas cards register \
  --domain ri.kinetic.session.integrity \
  --app-family ri.ecology \
  --family-class digital_information \
  --authored-by "Mycal Brooks" \
  --label "RI Kinetic Session Integrity" \
  --workflow session_integrity \
  --pulse-type pgdag

# import/export raw card JSON
sbn atlas cards import frontier-card.json
sbn atlas cards import frontier-card.json --supersede-existing --change-note "RI refinement"
sbn atlas cards export ri.kinetic.session.integrity --out ri-card-export.json

# attach a card to an app workflow and optionally bind an existing key
sbn atlas cards attach ri.kinetic.session.integrity \
  --app-id <app-id> \
  --workflow session_integrity \
  --key-id <existing-api-key-id>
```

`cards import` accepts either:

- an SBN-exported frontier card read model from `cards export`
- or a raw RI-authored JSON card payload centered on `frontier_card`

That makes it practical to author cards in another system, then bring them into
SBN for inspection, testing, simulation, and later workflow binding.

A neutral starter template is available at:

- [ATLAS_FRONTIER_CARD_TEMPLATE.json](/C:/Users/mycal/tower_projects/smartblocks_v2/docs/catalogue/ATLAS_FRONTIER_CARD_TEMPLATE.json)

## Lattice-backed pgDAG execution

When a workflow lives in Lattice, the SDK can load it as a `pgdag.DAGTemplate`
with the linked `workflow_id` already embedded in template metadata. That means
pgDAG execution can emit proof-aware Lattice outcomes automatically as steps seal
and publish.

```python
from sbn import SbnClient
from pgdag import build, StepContext

client = SbnClient(base_url="https://api.smartblocks.network")
client.authenticate_api_key("sbn_live_abc123")

layer = build(
    sbn_client=client,
    lattice_client=client.lattice,
    domain="kinetic.session",
)

template = client.lattice.build_pgdag_template("wf_kinetic_session_integrity_v3_ab12cd34")

async def collect_signal(ctx: StepContext) -> dict:
    return {"signal": "collected", "sample_id": ctx.inputs["sample_id"]}

async def seal_integrity(ctx: StepContext) -> dict:
    return {"integrity": "verified", "sample_id": ctx.inputs["sample_id"]}

layer.runner.register_steps(
    {
        "collect_signal": collect_signal,
        "seal_integrity": seal_integrity,
    }
)

result = await layer.runner.execute(
    template,
    inputs={"sample_id": "sess-42"},
    actor="kinetic-integrity-engine",
)
```

CLI helpers:

```bash
sbn lattice workflows create \
  --name "kinetic.pushup.amrap.v1" \
  --domain kinetic.pushup \
  --nodes-file nodes.json \
  --edges-file edges.json \
  --meta-file runtime_binding.json \
  --activate

sbn lattice workflows import workflow-export.json --activate
sbn lattice workflows import workflow-bundle.json
sbn lattice workflows template <workflow-id>
sbn lattice workflows template <workflow-id> --output kinetic_session_template.json
sbn lattice workflows runs <workflow-id>
sbn lattice workflows evidence-packet <workflow-id> <run-id> --output run-packet.json
sbn lattice workflows verify-packet --file run-packet.json
```

Reality review workflows can also be handled directly from the CLI:

```bash
sbn reality facts review <fact-id> --action accept
sbn reality facts review <fact-id> --action merge --target-fact-id <existing-fact-id> --note "duplicate extraction"
```

