refactor(api-types): extract shared cloud wire contract from soth-sync

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "2"
 members = [
     "crates/soth-cli",
     "crates/soth-core",
+    "crates/soth-api-types",
     "crates/soth-bundle",
     "crates/soth-classify",
     "crates/soth-conformance-tests",
@@ -47,6 +48,7 @@ authors = ["Labterminal"]
 [workspace.dependencies]
 # Internal crates
 soth-core = { path = "crates/soth-core" }
+soth-api-types = { path = "crates/soth-api-types" }
 soth-bundle = { path = "crates/soth-bundle" }
 # Default features OFF at the workspace level so the WASM SDK target
 # can build without ONNX. Consumers that want local classification

bindings/soth-node/examples/batch_proof.js

@@ -0,0 +1,58 @@
+// Fire 20 cheap Anthropic calls concurrently through the Node SDK
+// and watch the shipper's POST lines so we can count how many HTTP
+// requests actually leave the box.
+//
+// Mirrors examples/batch_proof.py from soth-py — both bindings sit
+// on the same Rust shipper in soth-sdk-core, so the wire format and
+// batching cadence should be identical.
+
+const Anthropic = require('@anthropic-ai/sdk');
+const soth = require('..');
+
+if (!process.env.SOTH_ORG_ID || !process.env.SOTH_API_KEY) {
+  console.error('SOTH_ORG_ID and SOTH_API_KEY are required (find them in ~/.soth/soth.yaml)');
+  process.exit(1);
+}
+const ORG_ID = process.env.SOTH_ORG_ID;
+const SOTH_API_KEY = process.env.SOTH_API_KEY;
+const TELEMETRY_ENDPOINT = process.env.SOTH_TELEMETRY_ENDPOINT
+  || 'https://ingest.soth.ai/v1/edge/telemetry/batch';
+const MODEL = 'claude-haiku-4-5-20251001';
+const N_CALLS = 20;
+
+async function main() {
+  if (!process.env.ANTHROPIC_API_KEY) {
+    console.error('ANTHROPIC_API_KEY not set');
+    process.exit(1);
+  }
+
+  soth.init({
+    apiKey: SOTH_API_KEY,
+    orgId: ORG_ID,
+    telemetryEndpoint: TELEMETRY_ENDPOINT,
+  });
+  const state = soth.instrument({ providers: ['anthropic'] });
+  console.log('instrumentation:', state);
+
+  const client = new Anthropic.default();
+
+  const fire = async (i) => {
+    const msg = await client.messages.create({
+      model: MODEL,
+      max_tokens: 8,
+      messages: [{ role: 'user', content: `Reply with the number ${i}, nothing else.` }],
+    });
+    return msg.usage.output_tokens;
+  };
+
+  const t0 = Date.now();
+  const outs = await Promise.all(Array.from({ length: N_CALLS }, (_, i) => fire(i)));
+  const dt = (Date.now() - t0) / 1000;
+  console.log(`\n>>> fired ${N_CALLS} calls in ${dt.toFixed(2)}s, total output tokens: ${outs.reduce((a, b) => a + b, 0)}`);
+  console.log('>>> sleeping 12s to let shipper drain (BATCH_WINDOW=5s)…');
+  await new Promise((r) => setTimeout(r, 12_000));
+  soth.shutdown();
+  console.log('>>> shutdown complete (forces final-drain POST)');
+}
+
+main().catch((e) => { console.error(e); process.exit(1); });

bindings/soth-py/.gitignore

@@ -0,0 +1,21 @@
+# Python venvs created locally for `maturin develop`. The PyO3 build
+# happens against the active venv; CI uses a fresh one each run.
+.venv/
+venv/
+
+# Compiled native extension produced by `maturin develop`. CI builds
+# wheels via `maturin build` / `cibuildwheel`; nothing needs to be
+# committed here.
+python/soth/_soth_native*.so
+python/soth/_soth_native*.dylib
+python/soth/_soth_native*.pyd
+
+# Byte-compiled Python.
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Distribution artefacts.
+build/
+dist/
+*.egg-info/

bindings/soth-py/examples/anthropic_demo.py

@@ -0,0 +1,208 @@
+"""Run Anthropic agents through the soth-py SDK auto-instrumentation.
+
+Flow:
+    soth.init(api_key, org_id, telemetry_endpoint)  # cloud shipper
+    soth.instrument()                               # patch anthropic SDK
+    # ... normal anthropic.Anthropic() calls now run through SOTH's
+    #     pre/post lifecycle and ship telemetry to the configured
+    #     ingest endpoint, which is what the dashboard reads from.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+import time
+import uuid
+
+import anthropic
+import soth
+
+# Wire SOTH at the same org/api_key the local edge agent uses so the
+# SDK's telemetry lands in the workspace the dashboard renders. The
+# easy way is to read them from ~/.soth/soth.yaml; we keep the demo
+# config-free by requiring env vars the customer pastes once.
+ORG_ID = os.environ["SOTH_ORG_ID"]
+SOTH_API_KEY = os.environ["SOTH_API_KEY"]
+TELEMETRY_ENDPOINT = os.environ.get(
+    "SOTH_TELEMETRY_ENDPOINT", "https://ingest.soth.ai/v1/edge/telemetry/batch"
+)
+
+# Unique-per-run tag so we can find these events in the dashboard.
+RUN_TAG = f"soth-py-demo/{uuid.uuid4().hex[:8]}"
+MODEL = "claude-haiku-4-5-20251001"
+
+
+def section(title: str) -> None:
+    print()
+    print("=" * 72)
+    print(title)
+    print("=" * 72)
+
+
+def demo_oneshot(client: anthropic.Anthropic) -> None:
+    section("[1/3] one-shot")
+    t0 = time.time()
+    msg = client.messages.create(
+        model=MODEL,
+        max_tokens=128,
+        metadata={"user_id": RUN_TAG},
+        messages=[
+            {
+                "role": "user",
+                "content": "In one sentence: what does an L7 forward proxy do?",
+            }
+        ],
+    )
+    dt = time.time() - t0
+    text = "".join(b.text for b in msg.content if getattr(b, "type", None) == "text")
+    print(f"latency={dt*1000:.0f}ms  in={msg.usage.input_tokens}  out={msg.usage.output_tokens}")
+    print(text)
+
+
+def demo_agent(client: anthropic.Anthropic) -> None:
+    section("[2/3] agent loop with tool use")
+
+    def calc(expr: str) -> str:
+        allowed = set("0123456789+-*/.() ")
+        if not expr or set(expr) - allowed:
+            return f"refused: {expr!r}"
+        return str(eval(expr, {"__builtins__": {}}))  # noqa: S307
+
+    def weather(city: str) -> str:
+        fake = {"sf": "62F foggy", "nyc": "48F clear", "tokyo": "55F rain"}
+        return fake.get(city.lower().strip(), f"no data for {city}")
+
+    tools = [
+        {
+            "name": "calculator",
+            "description": "Evaluate a basic arithmetic expression.",
+            "input_schema": {
+                "type": "object",
+                "properties": {"expression": {"type": "string"}},
+                "required": ["expression"],
+            },
+        },
+        {
+            "name": "weather",
+            "description": "Look up weather for sf, nyc, or tokyo.",
+            "input_schema": {
+                "type": "object",
+                "properties": {"city": {"type": "string"}},
+                "required": ["city"],
+            },
+        },
+    ]
+
+    messages: list = [
+        {
+            "role": "user",
+            "content": (
+                "What's 47 * 19, and what's the weather in Tokyo? "
+                "Then say 'done.' on its own line."
+            ),
+        }
+    ]
+    for step in range(1, 6):
+        resp = client.messages.create(
+            model=MODEL,
+            max_tokens=512,
+            tools=tools,
+            metadata={"user_id": RUN_TAG},
+            messages=messages,
+        )
+        messages.append({"role": "assistant", "content": resp.content})
+        if resp.stop_reason != "tool_use":
+            text = "".join(b.text for b in resp.content if getattr(b, "type", None) == "text")
+            print(f"steps={step} stop={resp.stop_reason}")
+            print(text)
+            return
+        tool_results = []
+        for block in resp.content:
+            if getattr(block, "type", None) != "tool_use":
+                continue
+            if block.name == "calculator":
+                out = calc(block.input.get("expression", ""))
+            elif block.name == "weather":
+                out = weather(block.input.get("city", ""))
+            else:
+                out = f"unknown tool {block.name}"
+            tool_results.append(
+                {"type": "tool_result", "tool_use_id": block.id, "content": str(out)}
+            )
+        messages.append({"role": "user", "content": tool_results})
+    print("hit max_steps")
+
+
+def demo_stream(client: anthropic.Anthropic) -> None:
+    section("[3/3] streaming")
+    t0 = time.time()
+    chunks = 0
+    with client.messages.stream(
+        model=MODEL,
+        max_tokens=160,
+        metadata={"user_id": RUN_TAG},
+        messages=[
+            {
+                "role": "user",
+                "content": "List 3 fun facts about TLS in numbered bullets.",
+            }
+        ],
+    ) as stream:
+        for text in stream.text_stream:
+            chunks += 1
+            sys.stdout.write(text)
+            sys.stdout.flush()
+        final = stream.get_final_message()
+    dt = time.time() - t0
+    print()
+    print(
+        f"\nchunks={chunks} latency={dt*1000:.0f}ms "
+        f"in={final.usage.input_tokens} out={final.usage.output_tokens}"
+    )
+
+
+def main() -> int:
+    if not os.environ.get("ANTHROPIC_API_KEY"):
+        print("ANTHROPIC_API_KEY not set", file=sys.stderr)
+        return 1
+
+    # Verbose so we can see the telemetry shipper's POST results.
+    logging.basicConfig(level=logging.INFO, format="%(name)s %(levelname)s: %(message)s")
+
+    print(f"run tag: {RUN_TAG}")
+    print(f"telemetry endpoint: {TELEMETRY_ENDPOINT}")
+    print(f"org_id: {ORG_ID}")
+
+    soth.init(
+        api_key=SOTH_API_KEY,
+        org_id=ORG_ID,
+        telemetry_endpoint=TELEMETRY_ENDPOINT,
+    )
+    state = soth.instrument(providers=["anthropic"])
+    print(f"instrumentation: {state}")
+    print(f"is_instrumented(anthropic) = {soth.is_instrumented('anthropic')}")
+
+    client = anthropic.Anthropic()  # vanilla — instrumented in place
+
+    try:
+        demo_oneshot(client)
+        demo_agent(client)
+        demo_stream(client)
+    finally:
+        section("flushing telemetry…")
+        # Sleep > BATCH_WINDOW (5s) so the shipper drains, then shutdown
+        # forces a final flush.
+        time.sleep(7)
+        soth.shutdown()
+
+    section(
+        f"done — search dashboard for run tag '{RUN_TAG}' "
+        f"or org_id={ORG_ID}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

bindings/soth-py/examples/batch_proof.py

@@ -0,0 +1,60 @@
+"""Fire 20 cheap Anthropic calls concurrently through the SDK and
+print the shipper's POST lines so we can count how many HTTP requests
+actually leave the box.
+
+If batching works, 20 events should land in 1–2 POSTs (one if they
+all queue inside a single 5s batch window, two if some land after
+the first window fires).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import anthropic
+import soth
+
+ORG_ID = os.environ["SOTH_ORG_ID"]
+SOTH_API_KEY = os.environ["SOTH_API_KEY"]
+TELEMETRY_ENDPOINT = os.environ.get(
+    "SOTH_TELEMETRY_ENDPOINT", "https://ingest.soth.ai/v1/edge/telemetry/batch"
+)
+MODEL = "claude-haiku-4-5-20251001"
+N_CALLS = 20
+
+
+def fire(client: anthropic.Anthropic, i: int) -> int:
+    msg = client.messages.create(
+        model=MODEL,
+        max_tokens=8,
+        messages=[{"role": "user", "content": f"Reply with the number {i}, nothing else."}],
+    )
+    return msg.usage.output_tokens
+
+
+def main() -> int:
+    if not os.environ.get("ANTHROPIC_API_KEY"):
+        print("ANTHROPIC_API_KEY not set", file=sys.stderr)
+        return 1
+
+    soth.init(api_key=SOTH_API_KEY, org_id=ORG_ID, telemetry_endpoint=TELEMETRY_ENDPOINT)
+    soth.instrument(providers=["anthropic"])
+    client = anthropic.Anthropic()
+
+    t0 = time.time()
+    with ThreadPoolExecutor(max_workers=10) as ex:
+        outs = list(ex.map(lambda i: fire(client, i), range(N_CALLS)))
+    dt = time.time() - t0
+    print(f"\n>>> fired {N_CALLS} calls in {dt:.2f}s, total output tokens: {sum(outs)}")
+    print(">>> sleeping 12s to let shipper drain (BATCH_WINDOW=5s)…")
+    time.sleep(12)
+    soth.shutdown()
+    print(">>> shutdown complete (forces final-drain POST)")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())