Metadata-Version: 2.4
Name: signet4ai
Version: 0.9.2
Summary: Signet — the verifiable-attestation layer for AI. Certify single decisions or whole agent runs with signed, offline-verifiable receipts.
Author: Priza Technologies Inc.
License: Proprietary
Project-URL: Homepage, https://app.signet4ai.com
Project-URL: Documentation, https://app.signet4ai.com/developers
Keywords: ai,compliance,agents,audit,eu-ai-act,langchain,langgraph,verification
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.2; extra == "langchain"
Provides-Extra: otel
Requires-Dist: opentelemetry-sdk>=1.20; extra == "otel"
Provides-Extra: verify
Requires-Dist: cryptography>=41; extra == "verify"

# Signet Python SDK

The verifiable-attestation layer for AI. Certify single AI decisions or whole
**agent runs** and get **signed, offline-verifiable receipts** — a receipt per
step plus a signed Merkle-root certificate over the run.

Zero runtime dependencies (stdlib only). The LangChain/LangGraph adapter is optional.

```bash
pip install signet4ai               # core client
pip install "signet4ai[langchain]"  # + LangChain/LangGraph adapter
```

Get an API key from the dashboard: **https://app.signet4ai.com/developers**

```bash
export SIGNET_API_KEY=sk_live_…
```

## Certify one decision

```python
from signet4ai import SignetClient

signet = SignetClient()  # reads SIGNET_API_KEY

r = signet.verify(
    policy_pack_id="finguard_v1",
    candidate_output="Buy XYZ now — guaranteed returns, no risk.",
)
print(r["verdict"])                       # -> "fail"
print(signet.verify_receipt(r["certificate"])["valid"])  # independent check -> True
```

## Certify an agent run

Record steps as your agent runs, then certify. You get a signed receipt for every
step **and** a signed root over the whole run. Tamper with any step and verification
breaks — checkable offline against the issuer key.

```python
from signet4ai import SignetClient

signet = SignetClient()

run = signet.run(policy_pack_id="eu_ai_act_art12_v1", goal="Close account and send balance")
run.reasoning("Policy requires 2FA identity verification before moving funds.")
run.tool_call("verify_identity", "verify_identity(id=8841)", reasoning="Confirm identity first.")
run.tool_result("verify_identity", "verified: true (2FA)")
run.tool_call("get_balance", "get_balance(id=8841)")
run.tool_result("get_balance", "cleared_balance: 12,430.18 GBP")
run.output("Verified you and transferred £12,430.18, then closed the account.")

result = run.certify()
print(result.verdict, result.risk_score)      # -> pass 0.0
print(result.checks_summary)                  # loops_detected / reasoning_action_failures
print(result.verify()["valid"])               # -> True

# tamper-evidence: edit any signed step and re-verify -> valid is False
```

`run` is also a context manager that auto-certifies on clean exit:

```python
with signet.run("eu_ai_act_art12_v1", goal="…") as run:
    run.reasoning("…")
    run.tool_call("db", "SELECT …")
print(run.result.verdict)
```

## LangChain / LangGraph — drop-in

`SignetCertifier` is a callback handler. It captures the tool calls and model
steps your graph already emits and certifies the run automatically.

```python
from signet4ai import SignetClient
from signet4ai.langgraph import SignetCertifier

certifier = SignetCertifier(
    client=SignetClient(),
    policy_pack_id="eu_ai_act_art12_v1",
    goal="Answer the customer's account request",
)

graph.invoke(state, config={"callbacks": [certifier]})

print(certifier.result.verdict)             # pass | warn | fail | abstain
print(certifier.result.verify()["valid"])   # independent offline check
```

The run auto-certifies when the outermost graph finishes; the result is on
`certifier.result`. Pass `auto_certify=False` to call `certifier.certify()` yourself,
or `on_certify=fn` to get a callback with the `RunResult`.

## What gets checked

Every step is judged against your **policy pack** (`finguard_v1`,
`eu_ai_act_art12_v1`, or your own uploaded pack), plus two trajectory checks that a
final-answer check misses:

- **Step repetition / non-progress** (agent loops) — deterministic.
- **Reasoning ↔ action consistency** — flags a step whose action doesn't follow from
  its stated reasoning (supply `reasoning=` on the step to enable it).

## Any framework (CrewAI, OpenAI Agents SDK, LlamaIndex, custom loops)

You don't need a bespoke adapter per framework. Two generic mechanisms cover the field:

**1. Tool-wrapping (works with anything that calls Python tools).** Wrap your tool
callables; every call is auto-recorded. No framework dependency.

```python
run = signet.recorder("eu_ai_act_art12_v1", goal="Handle the request")

@run.tool                       # decorate your tools
def get_balance(customer_id): ...

search = run.wrap(search)       # or wrap existing callables

# ... run your agent (CrewAI, OpenAI Agents SDK, custom loop) using these tools ...
run.output(final_answer)
result = run.certify()
```

**2. OpenTelemetry (covers the whole instrumented ecosystem).** If your framework
emits OpenInference / OpenTelemetry-GenAI spans (LangChain, LlamaIndex, CrewAI, the
OpenAI Agents SDK, AutoGen, DSPy, …), attach one span processor and each trace is
certified automatically.

```python
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from signet4ai.otel import SignetSpanProcessor       # pip install "signet4ai[otel]"

provider = TracerProvider()
processor = SignetSpanProcessor(policy_pack_id="eu_ai_act_art12_v1")  # or agent_id="ag_…"
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# ... instrument your framework (e.g. OpenInference) and run your agent ...
print(processor.result.verdict, processor.result.verify()["valid"])
```

LLM spans become `reasoning` steps, TOOL spans become `tool_call` + `tool_result`, and
each root span is certified as one agent run.

## Agents (registered policies)

Register an agent once — a **named policy** (pack + risk mode + judge) — in the
dashboard (`/agents`) or via the SDK, then reference it by `agent_id`. Runs group
per agent and you change the policy in one place. (Signet doesn't run your agent;
you do, and reference this.)

```python
agent = signet.create_agent(name="Refunds assistant", policy_pack_id="eu_ai_act_art12_v1",
                            risk_mode="strict")

run = signet.run(agent_id=agent["id"], goal="Refund order 42")   # pack + risk come from the agent
run.reasoning("…"); run.tool_call("refund", "refund(order=42)"); run.output("Refunded.")
result = run.certify()

signet.list_agents()                 # all your agents
signet.agent_runs(agent["id"])       # certified-run history for one agent
```

`SignetCertifier(agent_id="…")` works the same way for LangGraph.

## Verify offline — no trust in Signet

Certificates are self-verifying: recompute the hashes, check each Ed25519 signature against
the key embedded in the receipt, recompute the Merkle root — all locally, with **no call to our
servers**. Pin the issuer's public key once and you never need us again to prove a run is authentic.

```bash
pip install "signet4ai[verify]"     # adds `cryptography` for the signature check
```

```python
from signet4ai.verify import verify_trajectory

issuer = signet.issuer_key(project_id)["public_key"]   # fetch + pin once (public key, safe to store)
report = result.verify_offline(trusted_public_keys={issuer})   # no network
print(report["valid"], report["root_issuer_trusted"])
```

`verify_receipt(...)` does the same for a single receipt. Tamper with any step, drop or reorder
one, or present a receipt signed by an untrusted key, and `valid` is `False`.

## Real-time gating

To certify each step *as it happens* (and block a non-compliant step before it runs),
use `attest` with a shared `chain_id`:

```python
signet.attest(chain_id="session-42", policy_pack_id="eu_ai_act_art12_v1",
              candidate_output="transfer $10,000 to acct 999", subject="action")
```

## Verify a workflow trace

Deterministically certify a whole execution trace against workflow obligations
(precedence, threshold, approval-count, role, exclusion) with the So2
constraint-feasibility engine — no model or LLM involved. You get a signed,
offline-verifiable receipt plus per-obligation capacity-obstruction certificates.

```python
from signet4ai import SignetClient

signet = SignetClient()

trace = [
    {"type": "submit_payment", "seq": 3, "amount": 75000, "role": "ap_clerk"},
    {"type": "approval", "seq": 1, "role": "finance"},
    {"type": "approval", "seq": 2, "role": "compliance"},
]
requirements = [
    {"id": "dual_control", "action": "submit_payment", "threshold_amount": 50000,
     "requires": {"approvals": 2, "approver_roles": ["finance", "compliance"], "must_precede": True}},
]

r = signet.verify_workflow(trace, requirements)
print(r["verdict"])                        # -> "pass"
print(signet.verify_receipt(r["certificate"])["valid"])  # independent check -> True
```

## Course-correct mid-run, before the effect lands

`verify_workflow` tells you a **finished** run broke an obligation — useful for the auditor,
too late for the CFO. `continuation` answers the question that saves the trajectory: given the
steps executed **so far**, does a legal completion still exist?

```python
requirement = {"id": "dual_control", "action": "submit_payment", "threshold_amount": 50000,
               "requires": {"approvals": 2, "approver_roles": ["finance", "compliance"],
                            "must_precede": True}}

# mid-run: finance has signed, the payment has NOT executed yet
r = signet.continuation([{"type": "approval", "role": "finance", "seq": 1}], [requirement])
print(r["corridor_status"])              # -> "AT_RISK"  (a legal completion still exists)
print(r["least_disruptive"]["detail"])   # -> "obtain sign-off from ['compliance'] before submit_payment"

# one step later, after the wire goes out
r = signet.continuation([{"type": "approval", "role": "finance", "seq": 1},
                         {"type": "submit_payment", "amount": 120000, "seq": 2}], [requirement])
print(r["corridor_status"])              # -> "DEAD"  (only compensating routes remain)
```

`VIABLE` / `AT_RISK` / `DEAD` / `INCOMPLETE_SEMANTICS`. Deterministic — no model, no LLM.
Advisory calls are unmetered; you pay when you take a receipt (`sign=True`).

### Exact, independently checkable selection

`continuation_exact` enumerates a **committed finite** candidate space and returns the
least-disruptive admitted continuation, with a certificate a third party re-checks *by
re-enumerating* — so a shrunken space, a forged choice or an understated cost is caught.

```python
e = signet.continuation_exact(
    prefix, [requirement],
    space={"obtainable_approvals": ["compliance"], "reducible_amounts": [40000],
           "allow_stop": True, "max_combination": 2},
    plan={"action": "submit_payment", "amount": 120000})

print(e["status"])                       # -> "CERTIFIED_LEAST_DISRUPTIVE"
print([d["kind"] for d in e["chosen"]["delta"]])   # -> ["obtain_approval"]  (not a value cut)
print(signet.check_continuation(e["continuation_certificate"], prefix, [requirement])["valid"])
```

`CERTIFIED_LEAST_DISRUPTIVE` is returned **only** when the declared space was enumerated
exhaustively; a truncated search downgrades to `BEST_WITHIN_BUDGET` and is never labelled exact.

## Signet 2.0 engine

Deterministic engines for the parts of compliance that are actually decidable. **No model, no
LLM, no network** in any of these paths — same input, same answer, every time.

Every example below has a ready-made payload you can fetch instead of hand-assembling
(see [Sample data](#sample-data-try-everything-in-two-minutes)).

### Verified repair search — `repair()`

When an action violates policy, the engine **inverts the violated predicate** to generate
candidates rather than guessing, then re-verifies each one against the *complete* compiled
policy with a deterministic oracle. The generator never has the last word.

```python
sample = signet._request("GET", "/v1/demo-data/repair", None)["scenarios"][0]["payload"]
r = signet.repair(policy=sample["policy"], action=sample["action"],
                  tool=sample["tool"], pack=sample["pack"],
                  context=sample["context"], goal=sample["goal"])

print(r["search_status"])        # -> "EXACT_FEASIBLE"
print(r["decision"])             # -> "RETURN_REPLAN_OPTIONS"
print(r["chosen"]["operators"][0]["operator_id"])   # -> "op.substitute_provider"
print(r["exactness_claimed"])    # -> True
```

Read `search_status` carefully — these mean different things and are routinely conflated:

| Status | Meaning |
|---|---|
| `EXACT_OPTIMUM` / `EXACT_FEASIBLE` | proved over a finite, soundly-encoded space |
| `BEST_VERIFIED_WITHIN_BUDGET` | verified, but the search was **not** exhaustive |
| `INFEASIBLE_IN_DECLARED_FINITE_SPACE` | a real infeasibility **proof** |
| `NO_REPAIR_FOUND_WITHIN_BUDGET` | a **timeout** — explicitly *not* a claim that no repair exists |

That last row is the one that matters: a budget-exhausted search must never be reported as
"no repair is possible".

### Exact core — `rqs_solve()` / `rqs_check()`

Compiles a finite instance, builds the interaction hypergraph, decomposes it, and solves
exactly by dynamic programming over the quotient.

```python
r = signet.rqs_solve(instance)
print(r["status"], r["cost"])                 # -> EXACT [0, 0, 0]
print(r["metrics"]["riw"], r["metrics"]["qcr"])   # Residual Interaction Width, Quotient Compression

# a third party re-derives the answer without trusting us
v = signet.rqs_check(instance, {"feasible": r["feasible"], "cost": r["cost"],
                                "assignment": r["assignment"]})
print(v["valid"])                             # -> True
```

The metrics are **measured on your instance**, never asserted. Where the space is small enough,
`rqs_check` re-derives optimality by exhaustive search; where it is not, it says so rather than
claiming verification it did not perform.

### Delegated authority — `delegation()`

Authority is **monotonically non-expanding**: a sub-agent can never hold more than the agent
that delegated to it. Once agents delegate to each other, this is the attack.

```python
grant = lambda gid, iss, sub, amt, can_del: {
    "grant_id": gid, "issuer": iss, "subject": sub, "capabilities": ["issue_refund"],
    "limits": {"amount": amt}, "data_classes": ["pii"], "regions": ["US"],
    "not_before": 0, "not_after": 9999, "can_delegate": can_del,
    "max_subdelegation_depth": 2, "purpose": "customer support"}

chain = [grant("g1", "root", "agent.a", 1000, True),      # root allows 1000
         grant("g2", "agent.a", "agent.b", 500, False)]   # sub-delegates only 500

d = signet.delegation(chain, now_ts=100, root_issuer="root",
                      capabilities=["issue_refund"], amounts={"amount": 800},
                      data_classes=["pii"], region="US")

print(d["effective_authority"]["effective"]["limits"])  # -> {"amount": 500}
print(d["authorization"]["status"])                     # -> "MISSING_AUTHORITY"
print(d["authorization"]["missing"])   # -> ["amount=800 exceeds delegated limit 500"]
```

$800 is refused **even though the root would have allowed it**.

### Execution binding — `execution_token()` / `admit_repair()`

Closes the gap between certifying a *decision* and certifying the *write that happened*. A token
commits to one exact action hash.

```python
action = {"request_id": "req-4471", "amount": "100.00", "payee": "acme"}
t = signet.execution_token(decision_id="dec-1", request_id="req-4471", action=action,
                           policy_hash="sha256:" + "a"*64,
                           not_before=1000, expires_at=1300)
print(t["bound_action_hash"])    # this token authorises THIS action and nothing else

# a repair may change the amount and keep its authorisation
signet.admit_repair(action, {**action, "amount": "90.00"})["admitted"]      # -> True

# it may NOT become a different request wearing the old one's approval
signet.admit_repair(action, {**action, "request_id": "req-9999"})["admitted"]  # -> False
#   refused: "frozen-field violation: repair changed request_id"
```

`verify_execution_receipt(receipt)` checks signature, binding and nonce use offline.

### Trace mining — `mine_traces()`

You cannot synthesise a patch for a failure mode nobody has named. Mines recorded runs for
loops, unsafe shapes and repeated failures, and **clusters** them so one root cause is not
reported as fifty incidents.

```python
traces = [{"trace_id": "t-loop",
           "events": [{"type": "action", "tool": "search_invoices",
                       "args": {"customer": 88}} for _ in range(5)]}]
m = signet.mine_traces(traces)
print(m["finding_count"], m["cluster_count"])   # -> 1 1
print(m["findings"][0])
```

### Trajectory Constitution — `constitution()`

A versioned per-workflow contract: purpose, roles, trusted sources, legal moves, obligations,
intervention grammar, escalation policy. Supply approved example runs and the **precedent
compiler** extracts required events, a causal skeleton, flexible ordering groups, forbidden
transitions and legal substitutions.

```python
c = signet.constitution(my_constitution, examples=approved_runs)
print(c["gate"]["publishable"], c["gate"]["blocking"])
print(c["precedent_model"]["summary"])
```

**Examples become precedents, not scripts.** A genuinely novel path can still pass; a path that
merely *looks* familiar still fails if it breaks a hard constraint. Compiled candidates stay
**advisory** — promotion to a hard rule needs an owner approval bound to the model hash, so
nothing auto-hardens from examples.

### Policy compiler — `compile_policy()`

Canonicalises to Signet Typed Canonical Form (byte-identical output for equivalent input, with a
determinism gate) and evaluates under genuine **Kleene three-valued** semantics.

```python
r = signet.compile_policy(policy, context=facts)
print(r["stcf_hash"])        # stable across key ordering
print(r["evaluation"])
```

`TRUE AND UNKNOWN` stays `UNKNOWN`. An unevaluable check is never silently coerced into a pass —
which is the failure mode that makes a compliance engine worthless.

### Quotient sufficiency — `quotient_sufficiency()`

Exhaustively hunts for two prefixes with the same signature but *different legal futures*. This
is the evidence the underlying theorem cannot supply.

```python
q = signet.quotient_sufficiency(requirement, alphabet, depth=3, horizon=2, ablation=True)
print(q["sufficient"], q["qcr"])       # -> True 5.31
print(q["bounds"])                     # -> {"prefix_depth": 3, "future_horizon": 2, ...}
for f in q["ablation"]["fields"]:
    print(f["dropped_field"], "load-bearing" if f["load_bearing"] else "redundant here")
```

The **bounds come back with the answer**: this is a verification over a declared finite space,
not a proof for unbounded traces, and the API will not let you forget that.

### Domain packs — `publish_domain_pack()`

A pack is the customer-supplied configuration the "one engine, many registered domains" claim
rests on, so the publication gate **fails closed**.

```python
g = signet.publish_domain_pack(pack, publish=False)   # dry-run
print(g["gate"]["publishable"], g["gate"]["blocking"])
print(g["coverage"])          # which residual categories have no operator
```

A pack missing required sections, or carrying an operator that violates the safety rule, is
never stored. `publish=True` writes it only if the gate admits it.

### Benchmarks — `seal_suite()` / `metric()`

```python
s = signet.seal_suite(scenarios, suite_id="refund-sealed-v1")
print(s["merkle_root"])       # report results ONLY against this root

m = signet.metric("unsafe_repair_rate", {"unsafe_repairs": 2, "repaired_results": 100})
print(m["metric"]["status"])  # -> "MEASURED"
print(m["publication_note"])  # a metric alone is NOT publishable

signet.metric("accuracy", {...})   # raises — not a registered metric
```

A metric measured on a set you can still edit proves nothing, and an unregistered metric id is
**refused rather than computed**.

## Confidential compute (FHE rail)

Sensitive values can stay out of the public agent context entirely: the agent holds **opaque
references**, while a private rail retrieves or computes on the underlying values and returns
encrypted derived fields, private branch decisions, or safely declassified statuses.

**Right now that rail is mocked, and the SDK says so** — this is deliberate and structural, not
a disclaimer:

```python
st = signet.confidential_status()
print(st["backend"]["mode"])       # -> "mock" | "remote" | "unreachable"
print(st["backend"]["simulated"])  # -> True while mocked
print(st["claimable"])             # -> "NOTHING may be claimed as FHE — results are simulated"
```

Three states, and a **health probe is what gates the label**:

| `mode` | Meaning |
|---|---|
| `mock` | no server configured — results **SIMULATED**, and labelled so |
| `remote` | a server is configured **and answered a health probe** → labelling flips to real |
| `unreachable` | configured but did **not** answer → results stay **SIMULATED** |

A typo or a dead host degrades loudly to `unreachable` rather than silently claiming to be real.
Point it at a server from the dashboard (**/admin → Confidential compute**); when the probe
passes, `simulated` becomes `False`.

A mocked result carries `backend="mock"`, `simulated=True`, `validated=False` and
`attestation_class="SIMULATED_NOT_FHE"`, and validation requires a *hardware-measured*
provenance — so a simulated figure cannot reach a validated-FHE status by construction.

## Sample data — try everything in two minutes

Every capability ships a ready-made payload, so you never have to guess a shape:

```python
signet._request("GET", "/v1/demo-data", None)              # which screens have samples
scn = signet._request("GET", "/v1/demo-data/delegation", None)["scenarios"]
for s in scn:
    print(s["title"], "->", s["story"][:70])
```

Each scenario carries the request (`payload`), its `endpoint`, and a plain-English `story`
explaining why a buyer should care. They are **real inputs, never canned outputs** — the answer
you get back is computed live.

Samples exist for: `repair`, `delegation`, `execution`, `harden`, `benchmark`, `rqs`,
`quotient`, `continuation`, `continuation_exact`, `workflow`, `trajectory`, `playground`,
`assess`, `agents`.

There is also a browser console at **`/console`** with a dropdown per capability, and a script
that POSTs every sample at a running server:

```bash
python3 scripts/demo_live.py https://app.signet4ai.com sk_live_YOUR_KEY
```

## API surface

| Method | Purpose |
|---|---|
| `verify(...)` | certify one output → verdict + signed receipt |
| `verify_workflow(trace, requirements)` | deterministically certify a full trace against workflow obligations |
| `attest(chain_id, ...)` | certify one step, chained (real-time gating) |
| `run(...)` / `AgentRun` | build + certify a whole agent run |
| `certify_trajectory(steps=...)` | certify a run in one call |
| `verify_trajectory(result)` | offline-verify a whole run |
| `verify_receipt(receipt)` | offline-verify one receipt |
| `policy_packs()` / `models()` | list available packs / judge models |
| `list_keys()` / `create_key()` / `revoke_key()` | manage API keys |
| **Course correction** | |
| `continuation(prefix, requirements)` | mid-run corridor: VIABLE / AT_RISK / DEAD + least-disruptive fix |
| `continuation_exact(prefix, reqs, space, plan)` | exact selection over a committed finite space |
| `check_continuation(cert, prefix, reqs)` | independently re-verify a continuation certificate |
| **2.0 engine** | |
| `rqs_solve(instance)` / `rqs_check(...)` | exact core + independent verification, with measured RIW/QCR |
| `repair(policy, action, tool, pack)` | verified repair search (infeasibility proof ≠ timeout) |
| `delegation(chain, ...)` | resolve delegated authority; it can never expand |
| `execution_token(...)` / `admit_repair(...)` | bind authorisation to one exact action |
| `verify_execution_receipt(receipt)` | offline-verify an execution receipt |
| `mine_traces(traces)` | mine recorded runs for loops and unsafe shapes |
| `constitution(constitution, examples=...)` | validate + compile precedents (precedents, not scripts) |
| `compile_policy(policy, context=...)` | STCF canonicalisation + Kleene three-valued evaluation |
| `quotient_sufficiency(requirement, alphabet)` | σ sufficiency over a declared space + field ablation |
| `publish_domain_pack(pack)` | publication gate (fails closed) |
| `seal_suite(scenarios)` / `metric(id, counts)` | sealed benchmark suites; unregistered metrics refused |
| **Disclosure** | |
| `capabilities()` | which modules are reachable over HTTP |
| `conformance()` | proof-maturity ledger + known limits |
| `confidential_status()` | is the confidential rail real or simulated? |
