dealgo-csc 2.1.1

Python SDK for the DeAlgo decision firewall — runtime governance for autonomous AI agents

dealgo-csc

Runtime decision firewall for AI agents. The Python SDK for the DeAlgo decision API.

Wrap any model — trading strategy, LLM tool caller, agent — in client.decide(...), get back a binding permitted | blocked + reason, and respect the verdict. The system is deterministic, observable, and bounded. Every decision is hash-chained into a tamper-evident audit log you can verify byte-by-byte.

Measured result on a synthetic mean-reversion strategy across 5 seeds: ~8.5 percentage points drawdown reduction, robust 5/5. See dealgo-csc-lab/experiments/017_ab_synthetic for methodology.

CSC answers one question: Should this proposed action be permitted to execute right now?

It does not answer "is this profitable" or "is this optimal." It's the layer that stops a confused, hostile, or simply wrong model from firing at the worst time.

Why this exists

problem	what CSC does
Strategy fires bad-trade clusters during turbulent regimes	Token-bucket throttle structurally caps cluster volume → drawdown drops
LLM agent hallucinates a destructive tool call	Reality gate blocks; verdict carries the reason
Compromised agent fires actions in a flood	Bucket rate-limits at ~12 permits/min globally
Multi-agent system has conflicting goals	Per-source bias + safety judgment route the conflict
You want to learn from bad outcomes safely	v2 outcome channel: signed, integrity-gated, capped influence
Need post-hoc audit of every decision	Full pipeline trace + audit log per stimulus

Verified properties (lab phases 001–017)

• ~8.5 pp drawdown reduction robust across 5 seeds on cluster-prone mean-reversion strategy (Phase 017 multi-seed sweep)
• Zero verdict flips across 220+ real and synthetic decisions
• Deterministic safety geometry — same input, same verdict
• Clamped adaptive bias — v2 cannot flip a v1 verdict (capped ±0.05)
• Two-layer decay — old bias fades by both time AND traffic volume
• Byte-identical v1 path when callers don't opt into v2

What CSC is, and is not

	claim	evidence
defensible	"Reduces drawdown by suppressing loss clusters"	5/5 seeds, mean +8.5 pp drawdown improvement
❌	"Improves expectancy / makes strategies profitable"	tested, falsified — single-seed result was noise
❌	"Predicts better trades"	not a prediction system; pure control layer

What CSC actually does (verified, not marketing)

The SDK talks to the DeAlgo portal at https://api.dealgo.io/v1/decide, which runs the proposed action through a 3-layer governance pipeline. Each layer was characterized empirically in the lab (see Behavioral characterization below).

flowchart TD
    A([Agent proposes action]) --> SDK[client.decide]
    SDK -->|HTTPS POST /v1/decide<br/>Bearer sk_live_ or sk_test_| API[DeAlgo portal]

    API --> V{Validation boundary}
    V -- bad type / out of range / missing fields --> R422([HTTP 422<br/>raises ValidationError])

    V -- valid --> SPS{SPS gate<br/>survival policy}
    SPS -- DELAY --> Bp([blocked_policy])

    SPS -- APPROVE --> RG{Reality gate<br/>vol &lt; 0.15<br/>conf &gt;= 0.90}
    RG -- fail --> Br([blocked_reality_mismatch])

    RG -- pass --> TB{Token bucket<br/>capacity 4<br/>refill ~5s}
    TB -- empty --> Bt([blocked_controlled])
    TB -- token available --> P([permitted])

    Bp --> D[Decision]
    Br --> D
    Bt --> D
    P --> D
    D --> SDK
    SDK --> Caller([Caller acts<br/>or logs block])

---

Install

pip install dealgo-csc

Requires Python ≥3.9. Single runtime dependency: requests.

---

Quick start (v2.1+) — recommended for new integrations

Get a sandbox key from your portal at https://api.dealgo.io/keys (test keys don't count toward your monthly quota), then:

from csc_sdk import DeAlgo

client = DeAlgo(api_key="dealgo_sk_test_…")

verdict = client.decide(
    action_type="send_email",
    risk_score=0.3,
    confidence=0.95,
    intent="reply to customer with quote",
    metadata={"agent_id": "support-bot-7"},
)

if verdict.permitted:
    send_the_email()
else:
    print(f"Blocked: {verdict.reason}")
    print(f"Trace: https://api.dealgo.io/decisions/{verdict.decision_log_id}")

The decision_log_id is your direct link to the full pipeline trace: every gate that ran, every threshold compared, every reality check, in order — with a cryptographic chain proof.

Idempotency

Pass the same idempotency_key on a retry to avoid double-metering:

verdict = client.decide(
    action_type="send_email",
    risk_score=0.3,
    confidence=0.95,
    idempotency_key="my-request-uuid",   # safe to retry
)
if verdict.idempotency_replayed:
    print("This was a replayed response — meter not bumped twice.")

Closing the loop with report_outcome()

After observing the real-world result of an action, sign and report it back to feed the v2 adaptive-bias layer:

client.report_outcome(
    stimulus_id=verdict.id,
    magnitude=0.7,           # outcome was beneficial
    confidence=0.95,
    signature="…ed25519 base64…",
    public_key="…base64…",
    source="support-bot-7",
    context_type="General Stimulus",
    outcome_observed_at="2026-04-30T12:00:00Z",
    is_final=True,
)

---

Backwards compatibility — v1.x / v2.0 callers

Existing trading code keeps working unchanged:

from csc_sdk import CSC

csc = CSC(base_url="http://localhost:8765")   # direct-runtime mode

decision = csc.check(
    feed="trading",
    symbol="BTC",
    volatility=0.05,
    confidence=0.95,
    v2_aware=True,
)

if decision.permitted:
    execute_trade()

CSC is now an alias for DeAlgo. The check() method internally translates the trading-coded fields into the neutral schema before the request goes out, so trading callers get identical behavior with no code changes.

v2_aware=True still folds per-source outcome history into the SPS score (capped ±0.05). decision.v2_bias is None when v2 didn't fire.

---

The Decision object

field	type	meaning
permitted	bool	True = action may run
gate	Gate	which CSC layer decided: permit, policy, reality, throttle, override, unknown
reason	str	human-readable, suitable for logs and audit notes
id	str	stimulus ID for cross-referencing CSC's audit log
raw	dict	full underlying response body (audit / debug)

Decision is a frozen dataclass — safe to log, hash, pass between threads.

Gate semantics

flowchart LR
    G[Gate] --> P[permit<br/>action allowed]
    G --> Po[policy<br/>SPS verdict was DELAY]
    G --> R[reality<br/>vol or conf out of safe band]
    G --> T[throttle<br/>token bucket empty]
    G --> O[override<br/>founder override engaged]
    G --> U[unknown<br/>future or unrecognized state]

    style P fill:#cfc,stroke:#0a0
    style Po fill:#fcc,stroke:#a00
    style R fill:#fcc,stroke:#a00
    style T fill:#ffc,stroke:#aa0
    style O fill:#fcc,stroke:#a00
    style U fill:#eee,stroke:#888

Why distinguish throttle from policy/reality? Throttle is transient — the bucket refills in seconds and the same action will be permitted shortly. Policy and reality blocks indicate the action itself is unsafe; retrying will not help.

---

Convenience APIs

check_or_raise — for tool wrappers

from csc_sdk import CSC, BlockedError, ThrottledError

csc = CSC()

@tool  # your agent framework's tool decorator
def execute_trade(symbol: str, side: str):
    csc.check_or_raise(
        feed="trading", symbol=symbol,
        volatility=current_vol(), confidence=signal_confidence(),
        source="agent-1",
    )
    # If we get here, CSC permitted the action.
    return broker.place_order(symbol, side)

ThrottledError is a subclass of BlockedError, so wrappers can choose to retry or fail loudly per gate.

check_with_retry — for transient throttle handling

decision = csc.check_with_retry(
    feed="trading", symbol="BTC",
    volatility=0.05, confidence=0.95,
    max_attempts=4, base_backoff=5.0,
)
# Retries only on Gate.THROTTLE. Policy/reality blocks return immediately.

Backoff is linear: base_backoff * attempt, calibrated to the bucket's ~5s saturation behavior.

---

Testing your agent without a CSC server

from csc_sdk import Gate
from csc_sdk.testing import FakeCSC, permitted, blocked

# Always permits
csc = FakeCSC()

# Replays canned decisions in order
csc = FakeCSC([
    permitted("first-call"),
    blocked(Gate.REALITY, "volatility above 0.15"),
    blocked(Gate.THROTTLE, "no tokens"),
])

result = my_agent.run(csc=csc)
# Inspect what the agent asked CSC about:
assert csc.calls[0]["feed"] == "trading"

FakeCSC implements the same surface as CSC (check, check_or_raise, check_with_retry) but never opens a socket.

---

Operational characteristics (measured, not promised)

These numbers come from the lab harness adversarial sweeps; see Behavioral characterization below for methodology.

dimension	value	notes
Permit window (vol)	≤ 0.14	reality-gate hard threshold
Permit window (conf)	≥ 0.90	reality-gate hard threshold
Bucket capacity	4 permits	exact, deterministic
Bucket refill	~5s wall-clock	saturates from 0→4 in ~5s of quiet
Steady-state ceiling	~12 permits/min	4 permits per ~20s under continuous load
Per-decision latency	2–3 s	single CSC server, serial processing
Concurrent requests	serialized	linear latency under contention
False permits under chaos	0 / 100	random uniform sampling
False permits under spike	0 / 18	adversarial vol/conf spikes
False permits under multi-agent	0 / 8	concurrent dangerous agents

---

Behavioral characterization

The companion lab harness (dealgo-csc-lab) ran 8 phases of black-box probing against CSC. Every characteristic above is grounded in one of them:

flowchart LR
    P1[Phase 1<br/>chaos baseline<br/>100/100 blocked] --> P2[Phase 2<br/>boundary sweep<br/>permit island found]
    P2 --> P3[Phase 3<br/>fine sweep<br/>blocked_controlled is stateful]
    P3 --> P4[Phase 4<br/>controlled gate probe<br/>permit runs always 4]
    P4 --> P5[Phase 5<br/>bucket characterization<br/>refill is wall-clock]
    P5 --> P6[Phase 6<br/>refill curve<br/>~5s saturation]
    P6 --> P7[Phase 7<br/>adversarial spike<br/>judgment correct]
    P7 --> P8[Phase 8<br/>multi-agent<br/>safety holds under concurrency]

---

Architecture (where this fits)

flowchart LR
    subgraph YOUR[Your application]
        AGENT[Agent / bot / workflow]
    end

    subgraph SDK[dealgo-csc-sdk this repo]
        CLIENT[CSC client]
        DECISION[Decision model]
        FAKE[FakeCSC for tests]
    end

    subgraph CORE[CSC server]
        VAL[Pydantic validation]
        SPS[SPS gate]
        REALITY[Reality validator]
        BUCKET[Execution gate<br/>token bucket]
        AUDIT[Audit log]
    end

    AGENT -->|csc.check| CLIENT
    CLIENT -->|HTTP POST| VAL
    VAL --> SPS
    SPS --> REALITY
    REALITY --> BUCKET
    BUCKET -->|writes| AUDIT
    BUCKET -->|response| CLIENT
    CLIENT -->|Decision| AGENT

This SDK is the only surface external code should touch. The CSC server is internal infrastructure. Treat the HTTP shape of /csc/live-stimulus as private — its full response contract may change without notice. The SDK pins the public contract.

---

Errors

flowchart TD
    E[CSCError] --> V[ValidationError<br/>HTTP 422 from server]
    E --> T[TransportError<br/>network or non-decision HTTP]
    E --> B[BlockedError<br/>raised by check_or_raise]
    B --> Th[ThrottledError<br/>specifically Gate.THROTTLE]

ValidationError.detail carries the Pydantic error list from the server, so callers can surface field-level diagnostics.

---

Running the SDK tests

pip install -e ".[test]"
pytest tests/ -q

8 unit tests cover Decision parsing across all gate types and FakeCSC's canned-response semantics.

---

Repo layout

dealgo-csc-sdk/
├── csc_sdk/
│   ├── __init__.py     public surface
│   ├── client.py       CSC HTTP client (check / check_or_raise / check_with_retry)
│   ├── models.py       Decision (frozen dataclass), Gate enum
│   ├── errors.py       exception hierarchy
│   └── testing.py      FakeCSC for offline tests
├── tests/              pytest suite
├── pyproject.toml
└── README.md           you are here

---

Three-repo system context

repo	role	contents
dealgo-csc	Core — runs the actual server	FastAPI app, SPS, reality validator, token bucket, audit
dealgo-csc-lab	Lab — black-box harness	Chaos / boundary / adversarial / multi-agent experiment runners
dealgo-csc-sdk (this)	SDK — public client	Typed CSC client, Decision, FakeCSC

The lab never imports core; the SDK never imports core. All three communicate only through the documented HTTP API.

---

Versioning

Current release: v2.1.0. Targets the public DeAlgo portal at https://api.dealgo.io/v1/decide. Compatible with the dealgo-csc runtime that backs that endpoint.

The DeAlgo.decide(...) shape and Decision fields are stable. The Gate enum may grow new variants — callers should handle Gate.UNKNOWN defensively (treat as block). The legacy CSC.check(...) trading-coded signature is preserved as a backwards-compat alias and auto-translates to the neutral schema.

See CHANGELOG.md for release notes.

---

License

Apache-2.0. See LICENSE (or the metadata on PyPI).