cb-conformance 1.0.0

Thin Python SDK for emitting decision-lineage from managed external agents to CriticalBridge Brain

cb-conformance

Thin Python SDK for emitting decision-lineage from a managed external agent to CriticalBridge Brain.

The agent's TypeScript conformance sidecar (packages/conformance-sidecar) derives latency / counters from the A2A task envelope. This SDK complements the sidecar — it adds an in-process hook for things the sidecar can't see: chain-of-thought, retrieved citations, tool sub-calls, HITL touchpoints, and the NAIC V2 fields the Brain records per decision. Records correlate to the sidecar's task envelope via signal_id.

Wire-shape stability

This SDK targets the RecordDecisionLineageInput shape defined in src/main/brain-coordinator/state/DecisionLineageStore.ts (plus RetrievedContextSource from src/types/decision-lineage.types.ts), and the response shape returned by PR 2b (#8402)'s POST /api/brain/agents/v1/entities/:entityId/agents/:agentId/decision-lineage route — including the agent-attested HMAC authenticityEnvelope.

Until PR 2b merges, this SDK can be installed for local development but cannot round-trip against a live Brain. If PR 2b's wire shape diverges before merge, re-derive RetrievedContextEntry / ModelInvocation / HITLTouchpoint from the merged source of truth.

Install (local dev)

pip install -e packages/clients/python-conformance            # sync only
pip install -e 'packages/clients/python-conformance[async]'   # + async (httpx)

The async extra brings in httpx>=0.27 for AsyncConformanceClient. Sync-only users (requests-only) don't pay the cost.

Production install path is out of scope for this PR (it'll ship via the internal package index once Phase 2 stabilizes).

Use

import os
from cb_conformance import ConformanceClient, ModelInvocation, RetrievedContextEntry

client = ConformanceClient(
    brain_url="https://brain.acme.criticalbridge.ai",
    entity_id="ent-acme",
    agent_id="mgmt-langchain-broker-xyz",
    registration_token=os.environ["REGISTRATION_TOKEN"],  # from sidecar handoff
)

response = client.record_decision_lineage(
    signal_id="sig-from-sidecar-task-abc",   # correlates to the sidecar's task envelope
    work_item_id="wi-12345",
    skill="summarize-tickets",
    inputs={"prompt": "..."},
    outputs={"summary": "..."},
    risk_tier="medium",                       # 'low' | 'medium' | 'high' | 'critical'
    model_invocation=ModelInvocation(
        provider="anthropic",
        model="claude-opus-4-7",
        tokens_in=4200,
        tokens_out=180,
    ),
    retrieved_context=[
        RetrievedContextEntry(
            type="document",                  # 'document' | 'database' | 'rules_engine' | 'api'
            source_id="kb-faq-1",
            retrieved_at="2026-06-26T12:00:00.000Z",
            source_name="FAQ v3",
            excerpt="...trimmed...",
            relevance_score=0.91,
        ),
    ],
    consumer_impacting=False,
)

# Server-stamped response — persist the envelope alongside the record
# for any future auditor-side verification.
print(response["decisionId"])
print(response["signalId"])
print(response["decidedAt"])
envelope = response["authenticityEnvelope"]
print(envelope["signedBy"], envelope["signature"])

signal_id is optional — when omitted (or None), PR 2b's route stamps a fallback (mgmt-signal-<entityId>-<epoch-ms>). Read the server-stamped value off response["signalId"].

Async usage

Modern asyncio-native frameworks (LangChain, LlamaIndex, AutoGen) should use AsyncConformanceClient so a lineage POST doesn't block the event loop.

pip install 'cb-conformance[async]'

import os
from cb_conformance import AsyncConformanceClient, ModelInvocation

async with AsyncConformanceClient(
    brain_url="https://brain.acme.criticalbridge.ai",
    entity_id="ent-acme",
    agent_id="mgmt-langchain-broker-xyz",
    registration_token=os.environ["REGISTRATION_TOKEN"],
) as client:
    response = await client.record_decision_lineage(
        work_item_id="wi-12345",
        skill="summarize-tickets",
        signal_id="sig-from-sidecar-task-abc",
        model_invocation=ModelInvocation(
            provider="anthropic",
            model="claude-opus-4-7",
            tokens_in=4200,
            tokens_out=180,
        ),
    )
    print(response["decisionId"], response["authenticityEnvelope"]["signature"])

The async surface mirrors the sync ConformanceClient: same kwargs, same response shape, same exceptions, same retry behavior (250ms / 500ms / 1000ms + Retry-After honor, just asyncio.sleep instead of time.sleep). The underlying httpx.AsyncClient is created in __aenter__ and closed in __aexit__; connection pooling is automatic within the async with block.

Marshal + parse + retry-schedule helpers in _http.py are shared between the sync and async clients so the wire shape can't drift.

Behavior

• Sync + async. ConformanceClient (one requests.Session per instance, connection reuse across calls) and AsyncConformanceClient (one httpx.AsyncClient per async with block, async-native via httpx). Same surface, same wire shape, same retry behavior.
• Retry on 5xx. Up to 3 retries after the first attempt. Default backoff is exponential (250ms / 500ms / 1000ms). If a 5xx response carries a Retry-After: <seconds> header (PR 2b's dormant-flag 503 sends Retry-After: 60), the SDK honors it for the next attempt, clamped to [1.0, 300.0] seconds so a misconfigured server can't stall a calling agent indefinitely.
• 4xx is not retried. It's a caller bug — the Brain validated and said no.
• Redacted logs. On 4xx the SDK emits a single WARN with the status and signal_id. The bearer token and request body are never logged.
• Client-side validation. work_item_id and skill are required. risk_tier and data_classification, when present, are bounded by the same unions the TS route enforces.
• Wire shape. SDK surface is snake_case; the marshal layer converts to camelCase at the HTTP boundary (work_item_id → workItemId, model_invocation.tokens_in → modelInvocation.tokensIn, retrieved_context[].source_id → retrievedContext[].sourceId).
• Injectable sleep. ConformanceClient(sleep=...) accepts any Callable[[float], None] — tests use sleep=fake_sleeps.append to assert retry schedules without monkey-patching module attributes or waiting in real time.

Tests

cd packages/clients/python-conformance
python -m venv .venv && source .venv/bin/activate
pip install -e '.[dev]'   # includes async deps + pytest-asyncio
pytest                    # sync + async suite (~48 tests)
mypy --strict src/cb_conformance
ruff check src tests

Issue & epic

• Issue: #8378 (Phase 2 PR 2d — sync client)
• Epic: #8375 (Managed External Agents — async follow-up W8-PYTHON-ASYNC ships AsyncConformanceClient)
• Route PR (gates live round-trip): #8402 (Phase 2 PR 2b)