Metadata-Version: 2.4
Name: ulmp
Version: 0.1.1
Summary: Python client for UMP (Ulmen Message Protocol)
Author: El Mehdi Makroumi
License: BUSL-1.1
Requires-Python: >=3.10
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM

# ulmp

Python client for **UMP** (Ulmen Message Protocol) -- the wire protocol for agentic AI database workloads.

Backed entirely by a Rust implementation via PyO3. Zero protocol logic duplicated in Python. Maximum performance. Full type safety.

## Install

```bash
pip install ulmp
```

Requires Python 3.10+. The package includes a compiled Rust extension, no Rust toolchain needed to install.

## 30-second quickstart
Start the built-in test server:

```bash
python -m ulmp.testserver --port 7771 --token mytoken
```

In another terminal:

```python
import asyncio
from ulmp import AsyncClient

async def main():
    db = await AsyncClient.connect("localhost", 7771, token=b"mytoken")
    ns = db.namespace("my_project")

    await ns.put("auth.py::AuthService", b"class AuthService: pass")
    value = await ns.get("auth.py::AuthService")
    print(value)  # b"class AuthService: pass"

    await db.close()

asyncio.run(main())
```
That is it. You are connected, authenticated over TLS, and reading/writing records.

## Connection
``` python
from ulmp import AsyncClient

# Basic connection
db = await AsyncClient.connect(
    host="localhost",
    port=7771,
    token=b"your_secret_token",
)

# With all options
db = await AsyncClient.connect(
    host="db.internal",
    port=7771,
    token=b"secret",
    server_name="db.internal",      # TLS certificate name
    client_name="my-agent/1.0",     # identifies your client
    verify_cert=True,               # verify server certificate
    timeout=10.0,                   # connection timeout in seconds
)

# Always close when done
await db.close()
```

Use as an async context manager for automatic cleanup:
``` python
async with await AsyncClient.connect("localhost", 7771, token=b"secret") as db:
    ns = db.namespace("my_project")
    await ns.put("key", b"value")
    # connection closed automatically on exit
```

Check connection health:
```python
latency_us = await db.ping()        # round-trip in microseconds
print(db.is_connected)              # True
print(db.info.server_name)          # "ulmendb/0.1.0"
print(db.info.session_id)           # server-assigned session ID
```

## Namespaces

A namespace is a scoped view into one indexed codebase. All record operations are scoped to a namespace.

``` Python
ns = db.namespace("github.com/org/repo@abc123")
```

### Write
```Python
await ns.put("auth.py::AuthService", b"class AuthService: pass")
await ns.put("auth.py::validate_token", b"def validate_token(t): ...")
```

### Read
```Python
value = await ns.get("auth.py::AuthService")
# Returns bytes if found, None if not found
if value is None:
    print("key not found")
```

### Delete

``` Python
await ns.delete("auth.py::old_function")
```

### Batch write
```Python
await ns.put_batch([
    ("auth.py::AuthService", b"class AuthService: ..."),
    ("auth.py::validate_token", b"def validate_token(): ..."),
    ("auth.py::hash_password", b"def hash_password(p): ..."),
])
```

### Range scan
``` Python
# All keys starting with "auth.py::"
pairs = await ns.scan("auth.py::", "auth.py::zzz", limit=100)
for key, value in pairs:
    print(key, len(value), "bytes")
``` 

### Range delete
```Python
# Delete all keys in auth.py in one operation
await ns.range_delete("auth.py::", "auth.py::zzz")
```

## Query builder
Build queries with a fluent chaining API. The server runs all matching indices in parallel and merges results.

``` Python
results = await ns.query("JWT authentication validation") \
    .top_k(10) \
    .lang("python") \
    .type("function") \
    .relations("calls", "imports") \
    .file("src/*.py") \
    .timeout(5000) \
    .execute()

for row in results:
    print(f"{row.key}  score={row.final_score:.4f}  rank={row.rank}")
```

### Stream results
For large result sets, stream results one at a time and stop early:

``` Python
async for row in ns.query("validate token").top_k(50).stream():
    print(row.key, row.final_score)
    if row.final_score < 0.5:
        break  # sends ENOUGH to server, stops streaming
```

### Query methods

| Method | Description |
| --- | --- |
| .top_k(n) | Maximum results (default: 10) |
| .lang("python", "typescript") | Filter by programming language |
| .type("function", "class") | Filter by node type |
| .file("src/*.py") | Filter by file glob pattern |
| .relations("calls", "imports") | Follow relation edges in the graph |
| .depth(n) | Max graph traversal depth (default: 3) |
| .vector([0.1, 0.2, ...]) | Pre-computed embedding vector |
| .timeout(ms) | Cancel if not done in N milliseconds |
| .merge_rrf() | Reciprocal Rank Fusion merge (default) |
| .merge_linear() | Linear combination merge |
| .merge_max() | Max score merge |
| .execute() | Run and return all results |
| .stream() | Run and yield results as async iterator |

## Branches
Branch and merge like git. Native to the protocol, not an application-level hack.

``` Python
async with ns.branch("feat/refactor", description="refactor auth") as b:
    await b.put("auth.py::validate_token", new_implementation)
    await b.delete("auth.py::old_helper")
    # merges automatically on clean exit
    # rolls back automatically on any exception
```

If you need manual control:

``` Python
branch = ns.branch("experiment/risky-change")
async with branch as b:
    await b.put("auth.py::service", b"new code")
    # raise an exception here -> automatic rollback
    raise ValueError("this approach does not work")
    # branch is rolled back, original code is preserved
```

List and diff branches:

``` Python
branches = await ns.branch_list()
diff = await ns.branch_diff("feat/a", "feat/b")
```

## Transactions

ACID transactions with three isolation levels.

``` Python

async with db.transaction() as tx:
    val = await tx.get("counter")
    current = int(val or b"0")
    await tx.put("counter", str(current + 1).encode())
    # auto-commit on clean exit
    # auto-rollback on exception
```

Explicit control:
``` Python

async with db.transaction(isolation="serializable") as tx:
    await tx.put("key", b"value")
    await tx.commit()  # explicit commit

async with db.transaction() as tx:
    await tx.put("key", b"risky")
    await tx.rollback()  # explicit rollback
```

Isolation levels:

| Level | Description |
| --- | --- |
| "read_committed" | See committed writes from other transactions |
| "snapshot" | See a consistent snapshot from transaction start (default) |
| "serializable" | Full conflict detection, rejects on write-write conflict |

## Watch
Push notifications when keys change. No polling.

```Python
async with ns.watch("auth.py::") as stream:
    async for event in stream:
        print(f"{event.key} changed: {event.change_type}")
        # change_type: "put", "delete", "branch_create", "branch_merge"
```

## Snapshots
Create and restore point-in-time snapshots.

``` Python

# Save current state
await ns.snapshot_create("before refactor")

# List snapshots
snaps = await ns.snapshot_list()

# Restore to a previous state
await ns.snapshot_restore("snap-001")
```

## Distributed tracing
Every ULMP frame can carry a W3C-compatible trace context.

``` Python
from ulmp import TraceContext

# Root span
tc = TraceContext.generate()
print(tc.to_traceparent())  # "00-{trace_id}-{span_id}-01"

# Child span (same trace, new span)
child = tc.child_span()

# For custom protocol work
encoded = tc.encode()  # 25 bytes
```  

## Low-level API
For building raw UMP payloads when implementing custom protocol logic:

```Python
from ulmp import PayloadBuilder, decode_payload

# Build a payload
pb = PayloadBuilder()
pb.push_string("auth.py::AuthService")
pb.push_bytes(b"class AuthService: pass")
pb.push_u32(42)
pb.push_bool(True)
pb.push_null()
payload = pb.finish()  # bytes ending with TAG_END

# Decode a payload
values = decode_payload(payload)
# values = ["auth.py::AuthService", b"class AuthService: pass", 42, True, None]
```

### Frame encoding
``` Python

from ulmp import Header, HEADER_SIZE, MAGIC_V1

# Create a frame header
h = Header(opcode=0x06, stream_id=1, sequence=0, payload_length=9)
buf = h.encode()  # 20 bytes with CRC32

# Decode a frame header
h = Header.decode(buf)
print(h.opcode, h.stream_id, h.payload_length)
```

### Checkpoint tokens
For resuming long-running streaming operations after disconnect:

```Python

from ulmp import CheckpointToken

# Server creates a checkpoint
token = CheckpointToken(
    session_id=42,
    stream_id_high=0,
    stream_id_low=1,
    last_confirmed_row=1000,
    ttl_seconds=3600,
)
encoded = token.encode()  # 64 bytes

# Client decodes after reconnect
token = CheckpointToken.decode(encoded)
print(token.last_confirmed_row)  # 1000
print(token.is_expired())        # False
```

### Sequence tracking
Detect missing or duplicate frames:

``` Python

from ulmp import SequenceTracker

tracker = SequenceTracker()
assert tracker.check(0)   # True: first frame
assert tracker.check(1)   # True: sequential
assert not tracker.check(3)  # False: gap detected
print(tracker.expected)   # 2
```

## Crypto
All crypto runs in Rust at native speed. Zero external dependencies.

``` Python

from ulmp import crc32, sha256, hmac_sha256

# CRC32 (IEEE 802.3)
assert crc32(b"123456789") == 0xCBF43926

# SHA-256 (FIPS 180-4)
digest = sha256(b"hello world")
assert len(digest) == 32

# HMAC-SHA256 (FIPS 198-1)
mac = hmac_sha256(key=b"secret", msg=b"data")
assert len(mac) == 32
```

## Error handling
All errors have typed exceptions with error codes.

``` Python

from ulmp import (
    UlmpError,           # base exception
    ConnectionError,     # TCP/TLS failure
    AuthError,           # authentication rejected
    ProtocolError,       # frame/payload decode failure
    ServerError,         # server returned an error
    KeyNotFoundError,    # key does not exist
    NamespaceError,      # namespace not found or access denied
    QueryError,          # query execution failed
    TransactionError,    # transaction conflict or failure
    BranchError,         # branch not found or merge conflict
    RateLimitError,      # request rate exceeded
    TimeoutError,        # request deadline exceeded
)

# get() returns None for missing keys (no exception)
val = await ns.get("nonexistent")
assert val is None

# Server errors have codes
try:
    async with db.transaction(isolation="serializable") as tx:
        await tx.put("key", b"val")
except TransactionError as e:
    print(f"conflict: code=0x{e.code:02x}")
``` 

Exception hierarchy:

``` text
UlmpError
  ConnectionError
  AuthError
  ProtocolError
  ServerError
    KeyNotFoundError
    NamespaceError
    QueryError
    TransactionError
    BranchError
    RateLimitError
  TimeoutError
```

## Constants

All ULMP protocol constants are available from the Rust extension:

``` Python

from ulmp._core import (
    # Frame
    HEADER_SIZE,       # 20
    MAGIC_V1,          # 0x554D5001

    # 72 opcodes
    OP_PUT,            # 0x30
    OP_GET,            # 0x31
    OP_QUERY,          # 0x40
    OP_RESULT_ROW,     # 0x51
    OP_BRANCH_CREATE,  # 0x74

    # 13 capability bits
    CAP_TRACING,       # 1 << 3
    CAP_WATCH,         # 1 << 4
    ULMP_V1_CAPS,      # full v1 capability set

    # 31 error codes
    ERR_KEY_NOT_FOUND,  # 0x40
    ERR_TX_CONFLICT,    # 0x61

    # 8 flag bits
    FLAG_HAS_TRACE,     # 0b00000001
    FLAG_STREAM_END,    # 0b00010000
)
```

## Test server
The built-in test server lets you develop and test without a full ulmendb installation.

```Bash

# Start the test server
python -m ulmp.testserver --port 7771 --token mytoken
```

The test server:

- accepts TLS connections with a self-signed certificate
- authenticates via HMAC-SHA256 challenge-response
- stores records in memory (not persisted)
- supports PUT, GET, DELETE, SCAN, PING/PONG
- runs single-threaded, one connection at a time
Not for production use. Use ulmendb for production.

## Protocol
ulmp implements the UMP (Ulmen Message Protocol) specification:

- 72 opcodes across 16 groups
- 22-type self-describing payload system
- 20-byte frame header with CRC32 integrity
- TLS 1.3 mandatory, Ed25519 preferred
- HMAC-SHA256 challenge-response authentication
- Distributed tracing (W3C Trace Context compatible)
- Frame fragmentation for large payloads
- Connection-level and per-stream flow control
- In-session token rotation
- Checkpoint and stream resume
- Zero-downtime upgrade with drain/redirect
Full specification: [./docs]

## Architecture

``` text
Python layer (ulmp)           Rust core (ulmp._core)
  AsyncClient                   Frame encode/decode
  Namespace                     Payload type system
  QueryBuilder                  SHA-256, HMAC-SHA256, CRC32
  BranchContext                 TLS via rustls
  Transaction                   Checkpoint tokens
  WatchStream                   Sequence tracking
  Error hierarchy               72 opcode constants
  TestServer                    13 capability bits
                                31 error codes
```
All protocol logic runs in Rust. Python handles ergonomics and async I/O.

## Performance
The Rust extension handles all CPU-intensive work:

|Operation|Backend|
| --- | --- |
|Frame encode/decode|Rust|
|CRC32|Rust|
|SHA-256|Rust|
|HMAC-SHA256|Rust|
|Payload serialization|Rust|
|TLS handshake|Rust (rustls)|
|Async I/O|Python (asyncio)|

## Requirements

- Python 3.10+
- No external Python dependencies
- Rust extension included in wheel (no Rust toolchain needed to install)

## License
BSL 1.1

## Author
El Mehdi Makroumi

