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

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

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

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.

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

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).