From 8d570930d8f1f201e76e89ed134a0fe571ad1814 Mon Sep 17 00:00:00 2001 From: Apoorv Raj Saxena Date: Mon, 4 May 2026 17:02:00 +0530 Subject: [PATCH] refactor(api-types): extract shared cloud wire contract from soth-sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SDK and the proxy were silently drifting on the cloud wire format: the proxy converted in-process `soth_core::TelemetryEvent` -> wire `TelemetryEvent` via a private `map_event` in soth-sync, while the SDK shipper just `serde_json::json!`'d the raw in-process struct. That worked at first but blew up on the cloud as a 422 (`topic_cluster_id` expected string, got integer 0) — and any future field divergence would silently drop SDK telemetry without affecting proxy telemetry. Extract the cloud API types and the in-process -> wire conversion into a new pure-types crate `soth-api-types`. Both `soth-sync` (proxy) and `soth-sdk-core` (SDK) depend on it, so a contract change in either direction now propagates to both at once. Also fix the latent SDK shipper bug that produced the 422: - send `X-Soth-Api-Version: 2026-02-01` (matches soth-sync) - run each event through `soth_api_types::convert::map_event` - ship a real `TelemetryBatchRequest` envelope (batch_id, device_id_hash="sdk:", proxy_signature="sdk-unsigned"), matching the proxy envelope shape `SOTH_SHIPPER_VERBOSE=1` opts into stderr lines with the POST status + event count so SDK customers can verify batching cadence at runtime. Verified against the production cloud endpoint: 20 anthropic calls -> 1 POST /v1/edge/telemetry/batch -> 200 (20 events) …from both soth-py and soth-node, with `https_proxy` unset (SDK is genuinely proxy-independent). Build/test parity (CI mirror, locally green): - cargo fmt --all -- --check - cargo clippy --workspace --lib --bins --tests - cargo test --workspace --lib --bins - cargo test -p soth-conformance-tests --test parity (2/2) - WASM matrix: soth-detect / soth-classify on wasm32-{wasip1,unknown-unknown} - cargo build -p soth-py / soth-node - pytest bindings/soth-py/tests (45 passed, 2 skipped) - npm test in bindings/soth-node (32/32) Notes: * `observation_records` on `TelemetryBatchRequest` becomes `Option>` so soth-api-types stays free of the `soth-telemetry` (rusqlite/tokio) transitive closure. The proxy serializes its typed records to Value at the call site; JSON wire shape is identical. * `proxy_signature: "sdk-unsigned"` is a documented placeholder for the SDK lane — full ed25519 signing of SDK telemetry remains Phase-2 work, called out in shipper.rs. Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 11 + Cargo.toml | 2 + bindings/soth-node/examples/batch_proof.js | 58 ++ bindings/soth-py/.gitignore | 21 + bindings/soth-py/examples/anthropic_demo.py | 208 +++++++ bindings/soth-py/examples/batch_proof.py | 60 ++ crates/soth-api-types/Cargo.toml | 18 + crates/soth-api-types/src/api_types.rs | 574 ++++++++++++++++++++ crates/soth-api-types/src/convert.rs | 260 +++++++++ crates/soth-api-types/src/lib.rs | 22 + crates/soth-sdk-core/Cargo.toml | 5 +- crates/soth-sdk-core/src/shipper.rs | 147 +++-- crates/soth-sync/Cargo.toml | 1 + crates/soth-sync/src/api_types.rs | 573 +------------------ crates/soth-sync/src/telemetry/sender.rs | 296 +--------- 15 files changed, 1383 insertions(+), 873 deletions(-) create mode 100644 bindings/soth-node/examples/batch_proof.js create mode 100644 bindings/soth-py/.gitignore create mode 100644 bindings/soth-py/examples/anthropic_demo.py create mode 100644 bindings/soth-py/examples/batch_proof.py create mode 100644 crates/soth-api-types/Cargo.toml create mode 100644 crates/soth-api-types/src/api_types.rs create mode 100644 crates/soth-api-types/src/convert.rs create mode 100644 crates/soth-api-types/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index b9dc5fa9..466cd0e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4009,6 +4009,15 @@ dependencies = [ "sha1", ] +[[package]] +name = "soth-api-types" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "soth-core", +] + [[package]] name = "soth-bundle" version = "0.1.0" @@ -4362,6 +4371,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "soth-api-types", "soth-classify", "soth-core", "soth-detect", @@ -4391,6 +4401,7 @@ dependencies = [ "serde", "serde_json", "sha2", + "soth-api-types", "soth-bundle", "soth-core", "soth-telemetry", diff --git a/Cargo.toml b/Cargo.toml index b55bd249..65dc059e 100644 --- a/Cargo.toml +++ b/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 diff --git a/bindings/soth-node/examples/batch_proof.js b/bindings/soth-node/examples/batch_proof.js new file mode 100644 index 00000000..304f3575 --- /dev/null +++ b/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); }); diff --git a/bindings/soth-py/.gitignore b/bindings/soth-py/.gitignore new file mode 100644 index 00000000..488b3537 --- /dev/null +++ b/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/ diff --git a/bindings/soth-py/examples/anthropic_demo.py b/bindings/soth-py/examples/anthropic_demo.py new file mode 100644 index 00000000..96dce532 --- /dev/null +++ b/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()) diff --git a/bindings/soth-py/examples/batch_proof.py b/bindings/soth-py/examples/batch_proof.py new file mode 100644 index 00000000..48842a80 --- /dev/null +++ b/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()) diff --git a/crates/soth-api-types/Cargo.toml b/crates/soth-api-types/Cargo.toml new file mode 100644 index 00000000..d1db0bdd --- /dev/null +++ b/crates/soth-api-types/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "soth-api-types" +description = "SOTH cloud API wire contract — shared between proxy (soth-sync) and SDK (soth-sdk-core)." +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +authors.workspace = true + +# Pure types + a deterministic in-process → wire converter. No I/O, no +# tokio, no reqwest. Both the proxy and the SDK depend on this so they +# can never silently drift on the cloud contract. + +[dependencies] +soth-core = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } diff --git a/crates/soth-api-types/src/api_types.rs b/crates/soth-api-types/src/api_types.rs new file mode 100644 index 00000000..1958405e --- /dev/null +++ b/crates/soth-api-types/src/api_types.rs @@ -0,0 +1,574 @@ +//! Shared API request/response structures for sync <-> cloud communication. + +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, HashMap}; + +/// Header used for API version negotiation between edge and cloud. +pub const API_VERSION_HEADER: &str = "X-Soth-Api-Version"; + +/// Current API version expected by edge and cloud. +pub const API_VERSION: &str = "2026-02-01"; + +/// Compatibility submodule to preserve existing call sites. +pub mod version { + pub use super::{API_VERSION, API_VERSION_HEADER}; +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ExchangeBatchRequest { + pub agent_instance_id: String, + pub config_version: Option, + pub batch: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ExchangeMetadata { + pub exchange_id: String, + pub schema_version: String, + pub session_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub edge_session_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider_session_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_is_synthetic: Option, + pub observed_at: String, + pub started_at: Option, + pub completed_at: Option, + pub duration_ms: Option, + pub ttfb_ms: Option, + pub trace_id: Option, + pub span_id: Option, + pub parent_span_id: Option, + pub source_class: String, + pub transport: String, + pub provider: Option, + pub agent: Option, + pub model: Option, + pub endpoint: Option, + pub method: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detection_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detection_bundle_version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_identity_key: Option, + pub status_code: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub client_device_id: Option, + pub input_tokens: Option, + pub output_tokens: Option, + pub cache_read_tokens: Option, + pub cache_write_tokens: Option, + pub reasoning_tokens: Option, + pub cost_usd: Option, + pub cost_currency: Option, + pub pricing_version: Option, + pub request_size_bytes: Option, + pub response_size_bytes: Option, + pub request_body_mode: Option, + pub response_body_mode: Option, + pub request_body_ref: Option, + pub response_body_ref: Option, + pub request_body_sha256: Option, + pub response_body_sha256: Option, + pub request_body_preview: Option, + pub response_body_preview: Option, + pub request_truncated_reason: Option, + pub response_truncated_reason: Option, + pub truncated: bool, + pub metadata_only: bool, + pub discovery_capture: bool, + pub blacklist_match: bool, + pub pii_detected: bool, + pub pii_types: Vec, + pub policy_allowed: Option, + pub policy_version: Option, + pub mcp_tool_name: Option, + pub graphql_operation: Option, + pub event_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub integrity_status: Option, + pub signature: Option, + pub signature_key_id: Option, + pub parser_version: Option, + pub bundle_version: Option, + pub parse_confidence: Option, + pub detection_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detection_source: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub decision_step: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub decision_outcome: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub skip_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub discovery_kind: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub client_app_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub client_host_origin: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub client_referrer_origin: Option, + pub tags: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event_envelope: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventClientMetadata { + pub pid: Option, + pub device_id: Option, + pub bundle_id: Option, + pub process_name: Option, + pub process_executable: Option, + pub app_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub host_origin: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub referrer_origin: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventEnvelopeMetadata { + pub envelope_id: Option, + pub request_id: Option, + pub capture_source: Option, + pub source: Option, + pub captured_at: Option, + pub method: Option, + pub provider: Option, + pub host: Option, + pub path: Option, + pub model: Option, + pub agent: Option, + pub did: Option, + pub key_id: Option, + pub signature_alg: Option, + pub signed_fields_version: Option, + pub signature: Option, + pub body_hash: Option, + pub headers: Option>, + pub client: Option, + pub collector_source: Option, + pub collector_offset: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ExchangeBatchResponse { + pub accepted: u64, + pub rejected: u64, + pub errors: Vec, + #[serde(default)] + pub retry_after_secs: Option, + pub config_changed: bool, + pub server_time: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventError { + pub event_id: String, + pub reason: String, + #[serde(default)] + pub code: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BodyUploadResponse { + pub stored: bool, + pub request_key: Option, + pub response_key: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BlobUploadRequest { + pub exchange_id: String, + pub side: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub reference: Option, + pub content_encoding: Option, + pub content_type: Option, + pub sha256: Option, + pub bytes_raw: Option, + pub bytes_gzip: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub payload_gzip_b64: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BlobUploadResponse { + pub stored: bool, + #[serde(default)] + pub blob_key: Option, + #[serde(default)] + pub key: Option, + #[serde(default)] + pub sha256: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigResponse { + pub user: ConfigUser, + pub team: ConfigTeam, + pub org: ConfigOrg, + pub policies: Vec, + pub budget: ConfigBudget, + pub body_sync_level: String, + pub config_version: String, + #[serde(default)] + pub bundle_version: Option, + #[serde(default)] + pub registry_mode: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RegistryVersionResponse { + pub bundle_type: String, + pub version: String, + pub sha256: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bundle_hash: Option, + pub compiled_at: String, + #[serde(alias = "provider_count")] + pub llm_provider_count: u64, + pub domain_count: u64, + pub format_count: u64, + pub size_bytes: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub manifest: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub channel: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RegistryBundleManifest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bundle_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bundle_version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub published_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub diff_from: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub components: Vec, + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub changed_sections: HashMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub integrity: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RegistryBundleComponentHash { + pub name: String, + pub sha256: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub size_bytes: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RegistryBundleIntegrity { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub signature: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub signature_alg: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub key_id: Option, +} + +#[derive(Debug, Clone)] +pub enum RegistryBundleFetchQuery { + Full, + Section { + section: String, + }, + Diff { + from_hash: String, + section: Option, + }, +} + +impl Default for RegistryBundleFetchQuery { + fn default() -> Self { + Self::Full + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigUser { + pub id: String, + pub name: String, + pub email: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigTeam { + pub id: String, + pub name: String, + pub slug: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigOrg { + pub id: String, + pub name: String, + pub plan: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigPolicy { + pub name: String, + pub scope: String, + pub rego: String, + pub version: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigBudget { + pub enforcement: String, + pub limits: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConfigBudgetLimit { + pub scope: String, + pub model: Option, + pub daily_usd: Option, + pub weekly_usd: Option, + pub monthly_usd: Option, + pub remaining_usd: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HeartbeatRequest { + pub agent_instance_id: String, + pub proxy_version: String, + pub config_version: Option, + pub os: Option, + pub hostname: Option, + pub active_connections: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub host_details: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub registry: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub telemetry: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HeartbeatResponse { + pub ok: bool, + pub config_changed: bool, + pub server_time: String, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct HeartbeatTelemetry { + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub counters: BTreeMap, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct HeartbeatHostDetails { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub platform: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub os_family: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub os_version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub hostname: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub arch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cpu_logical_cores: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cpu_model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory_total_mb: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct HeartbeatRegistryDetails { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bundle_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bundle_version: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fetched_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bundle_age_seconds: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub validation_status: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub validation_failed_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub degraded_stale: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryBatchRequest { + pub batch_id: String, + pub org_id: String, + pub device_id_hash: String, + pub proxy_version: String, + pub timestamp: i64, + pub events: Vec, + pub proxy_signature: String, + /// Observation records from passive observer extensions. + /// Omitted when empty for backward compatibility. + /// + /// Stored as raw `serde_json::Value` so this crate stays free of + /// the `soth-telemetry` dep (and its rusqlite/tokio transitive + /// closure). The proxy serializes its typed + /// `soth_telemetry::ObservationTelemetryRecord` values into Values + /// at the call site; the SDK doesn't emit observations so it + /// always passes `None`. JSON wire shape is identical either way. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub observation_records: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryEvent { + pub event_id: String, + pub timestamp: i64, + pub provider: Option, + pub model: Option, + pub use_case_label: Option, + /// Why `use_case_label` has its current value. See + /// `soth_core::UseCaseLabelReason`. Serialized as snake_case string. + /// Skipped when omitted to keep older receivers backwards-compatible. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub use_case_label_reason: Option, + /// Top-1 model confidence in [0, 1]. None when not classified + /// (e.g. embedding skipped, fallback bundle). Lets the cloud filter + /// "uncertain" classifications and surface them for human review. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub use_case_confidence: Option, + /// Second-most-likely label when top-1 confidence < 0.40 — surfaces + /// multi-intent prompts the cloud can't currently see. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub secondary_label: Option, + pub topic_cluster_id: Option, + pub semantic_hash: Option, + #[serde(default)] + pub is_semantic_collision: bool, + pub collision_response_stability: Option, + pub anomaly_score: Option, + pub volatility_class: Option, + pub input_tokens: Option, + pub output_tokens: Option, + pub estimated_cost_usd: Option, + pub policy_decision: Option, + pub policy_rule_id: Option, + pub redaction_event: Option, + pub credential_pattern_detected: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detected_secret_types: Option>, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub detected_credential_types: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub languages: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub import_categories: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub auth_logic_detected: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub crypto_operations_detected: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub network_calls_detected: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub file_io_detected: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub private_key_detected: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub hardcoded_secret_detected: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub org_pattern_matches: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub anomaly_flags: Vec, + pub endpoint_hash: Option, + pub code_fraction: Option, + #[serde(default)] + pub tags: HashMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_key_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_prefix_repeat: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_code_context_repeat: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub novel_token_count: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub repeated_token_count: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub first_step_event_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub original_event_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub data_source: Option, + + // Caching intelligence fields + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dynamic_fraction: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub system_prompt_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_definition_hash: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prefix_repeat_signature: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub complexity_score: Option, + + // Response-side fields (populated when response data is available) + #[serde(default, skip_serializing_if = "Option::is_none")] + pub actual_output_tokens: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finish_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub response_latency_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ttfb_ms: Option, + + // Session metadata + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_request_count: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_total_tokens: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_credential_alerts: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub conversation_turn: Option, + + // WebSocket turn number + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ws_turn_number: Option, + + // Product/Session taxonomy (v7+) + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub product_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub surface_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_shadow_it: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub interaction_mode: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryBatchResponse { + pub accepted: u64, + pub rejected: u64, + pub errors: Vec, + pub config_changed: bool, + pub server_time: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelemetryBatchError { + pub event_id: String, + pub reason: String, +} diff --git a/crates/soth-api-types/src/convert.rs b/crates/soth-api-types/src/convert.rs new file mode 100644 index 00000000..03c79f00 --- /dev/null +++ b/crates/soth-api-types/src/convert.rs @@ -0,0 +1,260 @@ +//! In-process → wire conversion. +//! +//! `map_event` is the single canonical translator from the rich +//! in-process `soth_core::TelemetryEvent` to the cloud-bound +//! `api_types::TelemetryEvent`. The proxy (`soth-sync`) and the SDK +//! (`soth-sdk-core`) both call this so an envelope or field change +//! made here propagates to both sides at once. Drift here is what +//! produced the SDK→cloud `topic_cluster_id` 422 we hit before this +//! crate existed. + +use std::collections::HashMap; + +use soth_core::{ClassificationFlag, TelemetryPolicyKind, UseCaseLabelReason}; + +use crate::api_types::TelemetryEvent; + +pub fn map_event(event: &soth_core::TelemetryEvent) -> TelemetryEvent { + let mut tags = HashMap::new(); + let detected_credential_types = event.sensitive_code_flags.detected_secret_types.clone(); + if let Some(endpoint_type) = enum_name(&event.endpoint_type) { + tags.insert("endpoint_type".to_string(), endpoint_type); + } + if let Some(capture_mode) = enum_name(&event.capture_mode) { + tags.insert("capture_mode".to_string(), capture_mode); + } + if let Some(parse_confidence) = enum_name(&event.parse_confidence) { + tags.insert("parse_confidence".to_string(), parse_confidence); + } + if let Some(request_method) = enum_name(&event.request_method) { + tags.insert("request_method".to_string(), request_method); + } + if let Ok(parse_source_json) = serde_json::to_string(&event.parse_source) { + tags.insert("parse_source".to_string(), parse_source_json); + } + if let Some(bundle_trust_level) = event.bundle_trust_level.as_ref().and_then(enum_name) { + tags.insert("bundle_trust_level".to_string(), bundle_trust_level); + } + + // Emit app identity from process_resolution so cloud can group by tool + if let Some(ref pr) = event.process_resolution { + let source_class = match pr.app_type { + soth_core::AppType::NonHost => "agent_app", + soth_core::AppType::Host => "browser", + soth_core::AppType::Unknown => "unknown", + }; + tags.insert("source_class".to_string(), source_class.to_string()); + + let tool_key = pr + .matched_app_id + .as_deref() + .or(pr.process_name.as_deref()) + .or(pr.bundle_id.as_deref()); + if let Some(key) = tool_key { + tags.insert("tool_identity_key".to_string(), key.to_string()); + } + if let Some(ref name) = pr.process_name { + tags.insert("process_name".to_string(), name.clone()); + } + if let Some(ref bid) = pr.bundle_id { + tags.insert("bundle_id".to_string(), bid.clone()); + } + if let Some(match_kind) = enum_name(&pr.match_kind) { + tags.insert("match_kind".to_string(), match_kind); + } + if let Some(ref name) = pr.tool_name { + tags.insert("tool_name".to_string(), name.clone()); + } + if let Some(ref kind) = pr.tool_kind { + tags.insert("tool_kind".to_string(), kind.clone()); + } + if let Some(ref cat) = pr.tool_category { + tags.insert("tool_category".to_string(), cat.clone()); + } + if let Some(ref pid) = pr.provider_id { + tags.insert("provider_id".to_string(), pid.clone()); + } + } + + if let Some(ref ja4) = event.ja4_hash { + if !ja4.is_empty() { + tags.insert("ja4_hash".to_string(), ja4.clone()); + } + } + if let Some(ref tv) = event.tls_version { + if !tv.is_empty() { + tags.insert("tls_version".to_string(), tv.clone()); + } + } + if let Some(ref alpn) = event.alpn_protocol { + if !alpn.is_empty() { + tags.insert("alpn_protocol".to_string(), alpn.clone()); + } + } + if let Some(ref h2cid) = event.h2_connection_id { + if !h2cid.is_empty() { + tags.insert("h2_connection_id".to_string(), h2cid.clone()); + } + } + if let Some(h2sid) = event.h2_stream_id { + tags.insert("h2_stream_id".to_string(), h2sid.to_string()); + } + + // Historian-not-enriched warning is intentionally NOT emitted here + // (`soth-api-types` is a pure-types crate with no `tracing` dep). + // The proxy logs it at the call site that builds the batch; the + // SDK doesn't run historian enrichment so the case never fires. + let _ = matches!( + event.use_case_label_reason, + UseCaseLabelReason::HistorianNotEnriched + ); + + TelemetryEvent { + event_id: event.event_id.to_string(), + timestamp: event.timestamp_epoch_ms / 1_000, + provider: Some(event.provider.clone()), + model: event.model.clone(), + use_case_label: enum_name(&event.use_case), + use_case_label_reason: enum_name(&event.use_case_label_reason), + use_case_confidence: if event.use_case_confidence > 0.0 { + Some(event.use_case_confidence) + } else { + None + }, + secondary_label: event.secondary_label.as_ref().and_then(enum_name), + topic_cluster_id: if event.topic_cluster_id > 0 { + Some(event.topic_cluster_id.to_string()) + } else { + None + }, + semantic_hash: if event.semantic_hash.is_empty() { + None + } else { + Some(event.semantic_hash.clone()) + }, + is_semantic_collision: event.is_semantic_collision, + collision_response_stability: event.collision_response_stability.map(f64::from), + anomaly_score: event.anomaly_score.map(f64::from), + volatility_class: enum_name(&event.volatility_class), + input_tokens: event.estimated_input_tokens.map(u64::from), + output_tokens: event.estimated_output_tokens.map(u64::from), + estimated_cost_usd: event.estimated_cost_usd.map(f64::from), + policy_decision: event.policy_kind.as_ref().and_then(enum_name), + policy_rule_id: event.policy_rule_id.clone(), + redaction_event: Some(matches!( + event.policy_kind, + Some(TelemetryPolicyKind::Redact) + )), + credential_pattern_detected: Some( + event.sensitive_code_flags.credential_pattern_detected + || event + .classification_flags + .contains(&ClassificationFlag::CredentialDetected), + ), + detected_secret_types: if detected_credential_types.is_empty() { + None + } else { + Some(detected_credential_types.clone()) + }, + detected_credential_types, + languages: enum_names(&event.languages), + import_categories: enum_names(&event.import_categories), + auth_logic_detected: true_option(event.sensitive_code_flags.auth_logic_detected), + crypto_operations_detected: true_option( + event.sensitive_code_flags.crypto_operations_detected, + ), + network_calls_detected: true_option(event.sensitive_code_flags.network_calls_detected), + file_io_detected: true_option(event.sensitive_code_flags.file_io_detected), + private_key_detected: true_option(event.sensitive_code_flags.private_key_detected), + hardcoded_secret_detected: true_option( + event.sensitive_code_flags.hardcoded_secret_detected, + ), + org_pattern_matches: event.sensitive_code_flags.org_pattern_matches.clone(), + anomaly_flags: enum_names(&event.anomaly_flags), + endpoint_hash: if event.endpoint_hash.is_empty() { + None + } else { + Some(event.endpoint_hash.clone()) + }, + code_fraction: if event.code_fraction > 0.0 { + Some(f64::from(event.code_fraction)) + } else { + None + }, + tags, + session_key_hash: if event.session_key_hash.is_empty() { + None + } else { + Some(event.session_key_hash.clone()) + }, + is_prefix_repeat: if event.is_prefix_repeat { + Some(true) + } else { + None + }, + is_code_context_repeat: if event.is_code_context_repeat { + Some(true) + } else { + None + }, + novel_token_count: if event.novel_token_count > 0 { + Some(u64::from(event.novel_token_count)) + } else { + None + }, + repeated_token_count: if event.repeated_token_count > 0 { + Some(u64::from(event.repeated_token_count)) + } else { + None + }, + first_step_event_id: event.first_step_event_id.clone(), + original_event_id: event.original_event_id.clone(), + data_source: enum_name(&event.data_source), + dynamic_fraction: if event.dynamic_fraction > 0.0 { + Some(event.dynamic_fraction) + } else { + None + }, + system_prompt_hash: event.system_prompt_hash.clone(), + tool_definition_hash: event.tool_definition_hash.clone(), + prefix_repeat_signature: event.prefix_repeat_signature.clone(), + complexity_score: if event.complexity_score > 0 { + Some(event.complexity_score) + } else { + None + }, + actual_output_tokens: event.actual_output_tokens, + finish_reason: event.finish_reason.clone(), + response_latency_ms: event.response_latency_ms, + ttfb_ms: event.ttfb_ms, + session_request_count: event.session_request_count, + session_total_tokens: event.session_total_tokens, + session_credential_alerts: event.session_credential_alerts, + conversation_turn: event.conversation_turn, + ws_turn_number: event.ws_turn_number, + session_id: event.session_id.map(|u| u.to_string()), + product_id: event.product_id.clone(), + surface_type: enum_name(&event.surface_type), + is_shadow_it: if event.is_shadow_it { Some(true) } else { None }, + interaction_mode: enum_name(&event.interaction_mode), + } +} + +fn enum_name(value: &T) -> Option { + match serde_json::to_value(value).ok()? { + serde_json::Value::String(value) => Some(value), + _ => None, + } +} + +fn enum_names(values: &[T]) -> Vec { + values.iter().filter_map(enum_name).collect() +} + +fn true_option(value: bool) -> Option { + if value { + Some(true) + } else { + None + } +} diff --git a/crates/soth-api-types/src/lib.rs b/crates/soth-api-types/src/lib.rs new file mode 100644 index 00000000..68454cb5 --- /dev/null +++ b/crates/soth-api-types/src/lib.rs @@ -0,0 +1,22 @@ +//! SOTH cloud API wire contract. +//! +//! This crate is the single source of truth for the request/response +//! shapes the proxy (`soth-sync`) and the SDK (`soth-sdk-core`) speak +//! to the SOTH cloud. Keeping the types in one crate prevents the +//! kind of silent schema drift that would otherwise let one side +//! ship events the cloud rejects. +//! +//! - `api_types` — serde structs for telemetry, exchange, heartbeat, +//! config, registry, and blob upload endpoints. +//! - `convert` — `map_event` plus helpers that turn an in-process +//! `soth_core::TelemetryEvent` into the cloud-bound +//! `api_types::TelemetryEvent`. The proxy and the SDK both call +//! this so a wire-format change here propagates to both at once. +//! +//! No I/O, no tokio, no reqwest. WASM-compatible. + +pub mod api_types; +pub mod convert; + +pub use api_types::*; +pub use convert::map_event; diff --git a/crates/soth-sdk-core/Cargo.toml b/crates/soth-sdk-core/Cargo.toml index 185021d1..9a8e67bf 100644 --- a/crates/soth-sdk-core/Cargo.toml +++ b/crates/soth-sdk-core/Cargo.toml @@ -19,10 +19,13 @@ onnx-models = ["soth-classify/onnx-models"] otel = ["dep:opentelemetry", "dep:tracing-opentelemetry"] # Background HTTPS telemetry shipper. Off by default to keep workspace # `cargo build` fast; bindings flip it on for production builds. -http-telemetry = ["dep:reqwest"] +http-telemetry = ["dep:reqwest", "dep:soth-api-types"] [dependencies] soth-core = { workspace = true } +# Shared cloud wire contract. Optional + gated under `http-telemetry` +# so default builds (e.g. WASM) stay free of it. +soth-api-types = { workspace = true, optional = true } # soth-detect's workspace entry uses `default-features = false`; this crate # turns on `intelligence` so the in-memory backends are available, but # leaves `tree-sitter-code` and `intelligence-sqlite` off — neither is diff --git a/crates/soth-sdk-core/src/shipper.rs b/crates/soth-sdk-core/src/shipper.rs index fdbb75a9..7075331d 100644 --- a/crates/soth-sdk-core/src/shipper.rs +++ b/crates/soth-sdk-core/src/shipper.rs @@ -5,6 +5,11 @@ //! shipper is gated behind the `http-telemetry` feature so default //! builds stay light; bindings flip it on for production. //! +//! Wire format is the cloud-facing `soth_api_types::TelemetryBatchRequest`, +//! the same envelope the proxy ships via `soth-sync`. The SDK and the +//! proxy share a single source of truth for the cloud contract so +//! they cannot silently drift. +//! //! V0 deliberately ships a minimal shipper: bounded batches, fixed //! window, no retry/circuit-breaker. Phase 2 ports the full retry //! semantics from `soth-sync` (5 attempts with exp backoff, 72-hour @@ -17,6 +22,10 @@ use std::sync::Arc; use std::thread::JoinHandle; use std::time::Duration; +use soth_api_types::api_types::{TelemetryBatchRequest, API_VERSION, API_VERSION_HEADER}; +use soth_api_types::convert::map_event; +use soth_core::TelemetryEvent as CoreTelemetryEvent; + use crate::telemetry_queue::TelemetryQueue; const BATCH_WINDOW: Duration = Duration::from_secs(5); @@ -52,7 +61,6 @@ impl TelemetryShipper { pub fn shutdown(&mut self) { if !self.shutdown.swap(true, Ordering::Release) { if let Some(handle) = self.handle.take() { - // Best-effort join; ignore poisoned panics. let _ = handle.join(); } } @@ -101,55 +109,110 @@ fn run_loop( continue; } - // Wire envelope is intentionally minimal; cloud-side ingestion - // accepts the same shape soth-sync uses for the proxy. - let body = serde_json::json!({ - "org_id": &org_id, - "events": batch, - }); - - match client - .post(&endpoint) - .bearer_auth(&api_key) - .json(&body) - .send() - { - Ok(resp) if resp.status().is_success() => { - tracing::debug!( - target: "soth_sdk_core::shipper", - status = resp.status().as_u16(), - "telemetry batch posted" + post_batch( + &client, &endpoint, &api_key, &org_id, batch, /*final*/ false, + ); + } + + // Final drain on shutdown so events buffered during the last + // BATCH_WINDOW aren't lost on graceful exit. + let final_batch = queue.drain_batch(MAX_BATCH_SIZE); + if !final_batch.is_empty() { + post_batch( + &client, + &endpoint, + &api_key, + &org_id, + final_batch, + /*final*/ true, + ); + } +} + +fn post_batch( + client: &reqwest::blocking::Client, + endpoint: &str, + api_key: &str, + org_id: &str, + batch: Vec, + is_final: bool, +) { + let batch_size = batch.len(); + let request = build_request(org_id, batch); + let label = if is_final { "FINAL POST" } else { "POST" }; + + match client + .post(endpoint) + .header(API_VERSION_HEADER, API_VERSION) + .bearer_auth(api_key) + .json(&request) + .send() + { + Ok(resp) if resp.status().is_success() => { + tracing::debug!( + target: "soth_sdk_core::shipper", + status = resp.status().as_u16(), + "telemetry batch posted" + ); + if std::env::var("SOTH_SHIPPER_VERBOSE").is_ok() { + eprintln!( + "[soth-sdk-core::shipper] {} {} -> {} ({} events)", + label, + endpoint, + resp.status().as_u16(), + batch_size ); } - Ok(resp) => { - tracing::warn!( - target: "soth_sdk_core::shipper", - status = resp.status().as_u16(), - "telemetry POST returned non-success; events dropped (Phase-2 retry not yet wired)" + } + Ok(resp) => { + let status = resp.status().as_u16(); + let body_text = resp.text().unwrap_or_default(); + tracing::warn!( + target: "soth_sdk_core::shipper", + status, + "telemetry POST returned non-success; events dropped (Phase-2 retry not yet wired)" + ); + if std::env::var("SOTH_SHIPPER_VERBOSE").is_ok() { + eprintln!( + "[soth-sdk-core::shipper] {} {} -> {} ({} events dropped): {}", + label, endpoint, status, batch_size, body_text ); } - Err(error) => { - tracing::warn!( - target: "soth_sdk_core::shipper", - error = %error, - "telemetry POST error; events dropped" + } + Err(error) => { + tracing::warn!( + target: "soth_sdk_core::shipper", + error = %error, + "telemetry POST error; events dropped" + ); + if std::env::var("SOTH_SHIPPER_VERBOSE").is_ok() { + eprintln!( + "[soth-sdk-core::shipper] {} {} -> ERROR ({} events dropped): {}", + label, endpoint, batch_size, error ); } } } +} - // Final drain on shutdown so events buffered during the last - // BATCH_WINDOW aren't lost on graceful exit. - let final_batch = queue.drain_batch(MAX_BATCH_SIZE); - if !final_batch.is_empty() { - let body = serde_json::json!({ - "org_id": &org_id, - "events": final_batch, - }); - let _ = client - .post(&endpoint) - .bearer_auth(&api_key) - .json(&body) - .send(); +fn build_request(org_id: &str, batch: Vec) -> TelemetryBatchRequest { + let timestamp = batch.first().map(|e| e.timestamp_epoch_ms).unwrap_or(0); + let batch_id = uuid::Uuid::new_v4().to_string(); + let events = batch.iter().map(map_event).collect(); + + TelemetryBatchRequest { + batch_id, + org_id: org_id.to_string(), + // SDK has no signed device-id-hash flow; cloud accepts the + // bearer token + org_id alone for SDK-class telemetry. + device_id_hash: format!("sdk:{}", org_id), + proxy_version: format!("soth-sdk-core/{}", env!("CARGO_PKG_VERSION")), + timestamp, + events, + // Cloud requires a non-empty signature field but accepts the + // sentinel below for SDK-class submissions; full ed25519 + // signing is a Phase-2 deliverable. + proxy_signature: String::from("sdk-unsigned"), + observation_records: None, } } diff --git a/crates/soth-sync/Cargo.toml b/crates/soth-sync/Cargo.toml index a9f8f96b..9e0b715f 100644 --- a/crates/soth-sync/Cargo.toml +++ b/crates/soth-sync/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true [dependencies] soth-core = { workspace = true } +soth-api-types = { workspace = true } soth-bundle = { workspace = true } soth-telemetry = { workspace = true } anyhow = { workspace = true } diff --git a/crates/soth-sync/src/api_types.rs b/crates/soth-sync/src/api_types.rs index a834facd..a6a0a3e2 100644 --- a/crates/soth-sync/src/api_types.rs +++ b/crates/soth-sync/src/api_types.rs @@ -1,567 +1,14 @@ -//! Shared API request/response structures for sync <-> cloud communication. +//! Cloud API wire types — moved into the standalone `soth-api-types` +//! crate so the SDK (`soth-sdk-core`) and the proxy (`soth-sync`) +//! share one source of truth and cannot drift on the contract. +//! +//! This module re-exports the types under their original paths so +//! existing call sites elsewhere in `soth-sync` keep compiling. -use serde::{Deserialize, Serialize}; -use std::collections::{BTreeMap, HashMap}; +pub use soth_api_types::api_types::*; -/// Header used for API version negotiation between edge and cloud. -pub const API_VERSION_HEADER: &str = "X-Soth-Api-Version"; - -/// Current API version expected by edge and cloud. -pub const API_VERSION: &str = "2026-02-01"; - -/// Compatibility submodule to preserve existing call sites. +/// Compatibility submodule preserved for call sites that imported +/// `crate::api_types::version::*`. pub mod version { - pub use super::{API_VERSION, API_VERSION_HEADER}; -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ExchangeBatchRequest { - pub agent_instance_id: String, - pub config_version: Option, - pub batch: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ExchangeMetadata { - pub exchange_id: String, - pub schema_version: String, - pub session_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub edge_session_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub provider_session_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_is_synthetic: Option, - pub observed_at: String, - pub started_at: Option, - pub completed_at: Option, - pub duration_ms: Option, - pub ttfb_ms: Option, - pub trace_id: Option, - pub span_id: Option, - pub parent_span_id: Option, - pub source_class: String, - pub transport: String, - pub provider: Option, - pub agent: Option, - pub model: Option, - pub endpoint: Option, - pub method: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub detection_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub detection_bundle_version: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub tool_identity_key: Option, - pub status_code: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub client_device_id: Option, - pub input_tokens: Option, - pub output_tokens: Option, - pub cache_read_tokens: Option, - pub cache_write_tokens: Option, - pub reasoning_tokens: Option, - pub cost_usd: Option, - pub cost_currency: Option, - pub pricing_version: Option, - pub request_size_bytes: Option, - pub response_size_bytes: Option, - pub request_body_mode: Option, - pub response_body_mode: Option, - pub request_body_ref: Option, - pub response_body_ref: Option, - pub request_body_sha256: Option, - pub response_body_sha256: Option, - pub request_body_preview: Option, - pub response_body_preview: Option, - pub request_truncated_reason: Option, - pub response_truncated_reason: Option, - pub truncated: bool, - pub metadata_only: bool, - pub discovery_capture: bool, - pub blacklist_match: bool, - pub pii_detected: bool, - pub pii_types: Vec, - pub policy_allowed: Option, - pub policy_version: Option, - pub mcp_tool_name: Option, - pub graphql_operation: Option, - pub event_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub integrity_status: Option, - pub signature: Option, - pub signature_key_id: Option, - pub parser_version: Option, - pub bundle_version: Option, - pub parse_confidence: Option, - pub detection_reason: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub detection_source: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub decision_step: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub decision_outcome: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub skip_reason: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub discovery_kind: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub client_app_type: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub client_host_origin: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub client_referrer_origin: Option, - pub tags: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub event_envelope: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct EventClientMetadata { - pub pid: Option, - pub device_id: Option, - pub bundle_id: Option, - pub process_name: Option, - pub process_executable: Option, - pub app_type: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub host_origin: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub referrer_origin: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct EventEnvelopeMetadata { - pub envelope_id: Option, - pub request_id: Option, - pub capture_source: Option, - pub source: Option, - pub captured_at: Option, - pub method: Option, - pub provider: Option, - pub host: Option, - pub path: Option, - pub model: Option, - pub agent: Option, - pub did: Option, - pub key_id: Option, - pub signature_alg: Option, - pub signed_fields_version: Option, - pub signature: Option, - pub body_hash: Option, - pub headers: Option>, - pub client: Option, - pub collector_source: Option, - pub collector_offset: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ExchangeBatchResponse { - pub accepted: u64, - pub rejected: u64, - pub errors: Vec, - #[serde(default)] - pub retry_after_secs: Option, - pub config_changed: bool, - pub server_time: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct EventError { - pub event_id: String, - pub reason: String, - #[serde(default)] - pub code: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct BodyUploadResponse { - pub stored: bool, - pub request_key: Option, - pub response_key: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct BlobUploadRequest { - pub exchange_id: String, - pub side: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub reference: Option, - pub content_encoding: Option, - pub content_type: Option, - pub sha256: Option, - pub bytes_raw: Option, - pub bytes_gzip: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub payload_gzip_b64: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct BlobUploadResponse { - pub stored: bool, - #[serde(default)] - pub blob_key: Option, - #[serde(default)] - pub key: Option, - #[serde(default)] - pub sha256: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigResponse { - pub user: ConfigUser, - pub team: ConfigTeam, - pub org: ConfigOrg, - pub policies: Vec, - pub budget: ConfigBudget, - pub body_sync_level: String, - pub config_version: String, - #[serde(default)] - pub bundle_version: Option, - #[serde(default)] - pub registry_mode: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RegistryVersionResponse { - pub bundle_type: String, - pub version: String, - pub sha256: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub bundle_hash: Option, - pub compiled_at: String, - #[serde(alias = "provider_count")] - pub llm_provider_count: u64, - pub domain_count: u64, - pub format_count: u64, - pub size_bytes: u64, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub manifest: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub channel: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RegistryBundleManifest { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub bundle_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub bundle_version: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub published_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub diff_from: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub components: Vec, - #[serde(default, skip_serializing_if = "HashMap::is_empty")] - pub changed_sections: HashMap, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub integrity: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RegistryBundleComponentHash { - pub name: String, - pub sha256: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub size_bytes: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RegistryBundleIntegrity { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub signature: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub signature_alg: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub key_id: Option, -} - -#[derive(Debug, Clone)] -pub enum RegistryBundleFetchQuery { - Full, - Section { - section: String, - }, - Diff { - from_hash: String, - section: Option, - }, -} - -impl Default for RegistryBundleFetchQuery { - fn default() -> Self { - Self::Full - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigUser { - pub id: String, - pub name: String, - pub email: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigTeam { - pub id: String, - pub name: String, - pub slug: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigOrg { - pub id: String, - pub name: String, - pub plan: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigPolicy { - pub name: String, - pub scope: String, - pub rego: String, - pub version: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigBudget { - pub enforcement: String, - pub limits: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ConfigBudgetLimit { - pub scope: String, - pub model: Option, - pub daily_usd: Option, - pub weekly_usd: Option, - pub monthly_usd: Option, - pub remaining_usd: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HeartbeatRequest { - pub agent_instance_id: String, - pub proxy_version: String, - pub config_version: Option, - pub os: Option, - pub hostname: Option, - pub active_connections: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub host_details: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub registry: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub telemetry: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HeartbeatResponse { - pub ok: bool, - pub config_changed: bool, - pub server_time: String, -} - -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct HeartbeatTelemetry { - #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] - pub counters: BTreeMap, -} - -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct HeartbeatHostDetails { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub platform: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub os_family: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub os_version: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub hostname: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub arch: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub cpu_logical_cores: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub cpu_model: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub memory_total_mb: Option, -} - -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct HeartbeatRegistryDetails { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub bundle_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub bundle_version: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub fetched_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub bundle_age_seconds: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub validation_status: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub validation_failed_reason: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub degraded_stale: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TelemetryBatchRequest { - pub batch_id: String, - pub org_id: String, - pub device_id_hash: String, - pub proxy_version: String, - pub timestamp: i64, - pub events: Vec, - pub proxy_signature: String, - /// Observation records from passive observer extensions. - /// Omitted when empty for backward compatibility. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub observation_records: Option>, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TelemetryEvent { - pub event_id: String, - pub timestamp: i64, - pub provider: Option, - pub model: Option, - pub use_case_label: Option, - /// Why `use_case_label` has its current value. See - /// `soth_core::UseCaseLabelReason`. Serialized as snake_case string. - /// Skipped when omitted to keep older receivers backwards-compatible. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub use_case_label_reason: Option, - /// Top-1 model confidence in [0, 1]. None when not classified - /// (e.g. embedding skipped, fallback bundle). Lets the cloud filter - /// "uncertain" classifications and surface them for human review. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub use_case_confidence: Option, - /// Second-most-likely label when top-1 confidence < 0.40 — surfaces - /// multi-intent prompts the cloud can't currently see. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub secondary_label: Option, - pub topic_cluster_id: Option, - pub semantic_hash: Option, - #[serde(default)] - pub is_semantic_collision: bool, - pub collision_response_stability: Option, - pub anomaly_score: Option, - pub volatility_class: Option, - pub input_tokens: Option, - pub output_tokens: Option, - pub estimated_cost_usd: Option, - pub policy_decision: Option, - pub policy_rule_id: Option, - pub redaction_event: Option, - pub credential_pattern_detected: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub detected_secret_types: Option>, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub detected_credential_types: Vec, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub languages: Vec, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub import_categories: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub auth_logic_detected: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub crypto_operations_detected: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub network_calls_detected: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub file_io_detected: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub private_key_detected: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub hardcoded_secret_detected: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub org_pattern_matches: Vec, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub anomaly_flags: Vec, - pub endpoint_hash: Option, - pub code_fraction: Option, - #[serde(default)] - pub tags: HashMap, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_key_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub is_prefix_repeat: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub is_code_context_repeat: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub novel_token_count: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub repeated_token_count: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub first_step_event_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub original_event_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub data_source: Option, - - // Caching intelligence fields - #[serde(default, skip_serializing_if = "Option::is_none")] - pub dynamic_fraction: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub system_prompt_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub tool_definition_hash: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub prefix_repeat_signature: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub complexity_score: Option, - - // Response-side fields (populated when response data is available) - #[serde(default, skip_serializing_if = "Option::is_none")] - pub actual_output_tokens: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finish_reason: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub response_latency_ms: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub ttfb_ms: Option, - - // Session metadata - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_request_count: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_total_tokens: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_credential_alerts: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub conversation_turn: Option, - - // WebSocket turn number - #[serde(default, skip_serializing_if = "Option::is_none")] - pub ws_turn_number: Option, - - // Product/Session taxonomy (v7+) - #[serde(default, skip_serializing_if = "Option::is_none")] - pub session_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub product_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub surface_type: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub is_shadow_it: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub interaction_mode: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TelemetryBatchResponse { - pub accepted: u64, - pub rejected: u64, - pub errors: Vec, - pub config_changed: bool, - pub server_time: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TelemetryBatchError { - pub event_id: String, - pub reason: String, + pub use soth_api_types::api_types::{API_VERSION, API_VERSION_HEADER}; } diff --git a/crates/soth-sync/src/telemetry/sender.rs b/crates/soth-sync/src/telemetry/sender.rs index 1ba18f12..afd17deb 100644 --- a/crates/soth-sync/src/telemetry/sender.rs +++ b/crates/soth-sync/src/telemetry/sender.rs @@ -1,11 +1,9 @@ use anyhow::{Context, Result}; use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _}; use ed25519_dalek::{Signer, SigningKey}; -use soth_core::{ - derive_proxy_signing_seed, ClassificationFlag, TelemetryPolicyKind, UseCaseLabelReason, -}; +use soth_api_types::convert::map_event; +use soth_core::{derive_proxy_signing_seed, UseCaseLabelReason}; use soth_telemetry::{SignedBatch, TransmittedBatch}; -use std::collections::HashMap; use std::sync::Arc; use zeroize::Zeroizing; @@ -143,8 +141,34 @@ impl TelemetrySender { } fn signed_batch_to_request(&self, signed: &SignedBatch) -> Result { + // Mirror the historian-not-enriched WARN that used to sit inside + // `map_event`. It now lives at the call site so the shared + // `soth-api-types` crate stays free of a `tracing` dep. + for event in &signed.batch.events { + if matches!( + event.use_case_label_reason, + UseCaseLabelReason::HistorianNotEnriched + ) { + tracing::warn!( + event_id = %event.event_id, + "shipping historian event with use_case_label_reason=historian_not_enriched; \ + ClassifyEnricher likely failed or was skipped at ingest time" + ); + } + } + let events = signed.batch.events.iter().map(map_event).collect(); + // `soth-api-types::TelemetryBatchRequest.observation_records` is + // `Option>` so the shared crate doesn't + // pull in `soth-telemetry`. We serialize the typed records here. + let observation_records = signed.batch.observation_records.as_ref().map(|records| { + records + .iter() + .filter_map(|r| serde_json::to_value(r).ok()) + .collect::>() + }); + let mut request = TelemetryBatchRequest { batch_id: signed.batch.batch_id.to_string(), org_id: signed.batch.org_id.clone(), @@ -153,7 +177,7 @@ impl TelemetrySender { timestamp: signed.batch.timestamp_utc, events, proxy_signature: String::new(), - observation_records: signed.batch.observation_records.clone(), + observation_records, }; let signing_payload = TelemetrySigningPayload { @@ -182,268 +206,6 @@ struct TelemetrySigningPayload<'a> { events: &'a [TelemetryEvent], } -fn map_event(event: &soth_core::TelemetryEvent) -> TelemetryEvent { - let mut tags = HashMap::new(); - let detected_credential_types = event.sensitive_code_flags.detected_secret_types.clone(); - if let Some(endpoint_type) = enum_name(&event.endpoint_type) { - tags.insert("endpoint_type".to_string(), endpoint_type); - } - if let Some(capture_mode) = enum_name(&event.capture_mode) { - tags.insert("capture_mode".to_string(), capture_mode); - } - if let Some(parse_confidence) = enum_name(&event.parse_confidence) { - tags.insert("parse_confidence".to_string(), parse_confidence); - } - if let Some(request_method) = enum_name(&event.request_method) { - tags.insert("request_method".to_string(), request_method); - } - if let Ok(parse_source_json) = serde_json::to_string(&event.parse_source) { - tags.insert("parse_source".to_string(), parse_source_json); - } - if let Some(bundle_trust_level) = event.bundle_trust_level.as_ref().and_then(enum_name) { - tags.insert("bundle_trust_level".to_string(), bundle_trust_level); - } - - // Emit app identity from process_resolution so cloud can group by tool - if let Some(ref pr) = event.process_resolution { - // source_class: agent_app / browser / unknown - let source_class = match pr.app_type { - soth_core::AppType::NonHost => "agent_app", - soth_core::AppType::Host => "browser", - soth_core::AppType::Unknown => "unknown", - }; - tags.insert("source_class".to_string(), source_class.to_string()); - - // tool_identity_key: resolved app_id from detect bundle (e.g. "claude-code", "cursor") - // Fallback chain: matched_app_id → process_name → bundle_id - let tool_key = pr - .matched_app_id - .as_deref() - .or(pr.process_name.as_deref()) - .or(pr.bundle_id.as_deref()); - if let Some(key) = tool_key { - tags.insert("tool_identity_key".to_string(), key.to_string()); - } - if let Some(ref name) = pr.process_name { - tags.insert("process_name".to_string(), name.clone()); - } - if let Some(ref bid) = pr.bundle_id { - tags.insert("bundle_id".to_string(), bid.clone()); - } - if let Some(match_kind) = enum_name(&pr.match_kind) { - tags.insert("match_kind".to_string(), match_kind); - } - - // Unified registry resolved fields (v6+). - // Pre-resolved at edge so cloud can use directly without catalog lookup. - if let Some(ref name) = pr.tool_name { - tags.insert("tool_name".to_string(), name.clone()); - } - if let Some(ref kind) = pr.tool_kind { - tags.insert("tool_kind".to_string(), kind.clone()); - } - if let Some(ref cat) = pr.tool_category { - tags.insert("tool_category".to_string(), cat.clone()); - } - if let Some(ref pid) = pr.provider_id { - tags.insert("provider_id".to_string(), pid.clone()); - } - } - - // Connection intelligence tags (JA4 fingerprint, TLS metadata, H2 multiplexing) - if let Some(ref ja4) = event.ja4_hash { - if !ja4.is_empty() { - tags.insert("ja4_hash".to_string(), ja4.clone()); - } - } - if let Some(ref tv) = event.tls_version { - if !tv.is_empty() { - tags.insert("tls_version".to_string(), tv.clone()); - } - } - if let Some(ref alpn) = event.alpn_protocol { - if !alpn.is_empty() { - tags.insert("alpn_protocol".to_string(), alpn.clone()); - } - } - if let Some(ref h2cid) = event.h2_connection_id { - if !h2cid.is_empty() { - tags.insert("h2_connection_id".to_string(), h2cid.clone()); - } - } - if let Some(h2sid) = event.h2_stream_id { - tags.insert("h2_stream_id".to_string(), h2sid.to_string()); - } - - // Promote a one-time WARN when historian events bypass enrichment so we - // can spot misconfig at the egress edge (soth-core can't log here). - if matches!( - event.use_case_label_reason, - UseCaseLabelReason::HistorianNotEnriched - ) { - tracing::warn!( - event_id = %event.event_id, - "shipping historian event with use_case_label_reason=historian_not_enriched; \ - ClassifyEnricher likely failed or was skipped at ingest time" - ); - } - - TelemetryEvent { - event_id: event.event_id.to_string(), - timestamp: event.timestamp_epoch_ms / 1_000, - provider: Some(event.provider.clone()), - model: event.model.clone(), - use_case_label: enum_name(&event.use_case), - use_case_label_reason: enum_name(&event.use_case_label_reason), - // Tier A: previously dropped at egress. Only emit when classify - // produced a confidence (>0) — keeps payload size small for the - // many heuristic-parsed events that won't have a model output. - use_case_confidence: if event.use_case_confidence > 0.0 { - Some(event.use_case_confidence) - } else { - None - }, - secondary_label: event.secondary_label.as_ref().and_then(enum_name), - topic_cluster_id: if event.topic_cluster_id > 0 { - Some(event.topic_cluster_id.to_string()) - } else { - None - }, - semantic_hash: if event.semantic_hash.is_empty() { - None - } else { - Some(event.semantic_hash.clone()) - }, - is_semantic_collision: event.is_semantic_collision, - // Was hardcoded `None` here; now flows through from the in-process - // TelemetryEvent so any future upstream computation reaches the - // wire payload without another mapping change. - collision_response_stability: event.collision_response_stability.map(f64::from), - anomaly_score: event.anomaly_score.map(f64::from), - volatility_class: enum_name(&event.volatility_class), - input_tokens: event.estimated_input_tokens.map(u64::from), - output_tokens: event.estimated_output_tokens.map(u64::from), - estimated_cost_usd: event.estimated_cost_usd.map(f64::from), - policy_decision: event.policy_kind.as_ref().and_then(enum_name), - policy_rule_id: event.policy_rule_id.clone(), - redaction_event: Some(matches!( - event.policy_kind, - Some(TelemetryPolicyKind::Redact) - )), - credential_pattern_detected: Some( - event.sensitive_code_flags.credential_pattern_detected - || event - .classification_flags - .contains(&ClassificationFlag::CredentialDetected), - ), - detected_secret_types: if detected_credential_types.is_empty() { - None - } else { - Some(detected_credential_types.clone()) - }, - detected_credential_types, - languages: enum_names(&event.languages), - import_categories: enum_names(&event.import_categories), - auth_logic_detected: true_option(event.sensitive_code_flags.auth_logic_detected), - crypto_operations_detected: true_option( - event.sensitive_code_flags.crypto_operations_detected, - ), - network_calls_detected: true_option(event.sensitive_code_flags.network_calls_detected), - file_io_detected: true_option(event.sensitive_code_flags.file_io_detected), - private_key_detected: true_option(event.sensitive_code_flags.private_key_detected), - hardcoded_secret_detected: true_option( - event.sensitive_code_flags.hardcoded_secret_detected, - ), - org_pattern_matches: event.sensitive_code_flags.org_pattern_matches.clone(), - anomaly_flags: enum_names(&event.anomaly_flags), - endpoint_hash: if event.endpoint_hash.is_empty() { - None - } else { - Some(event.endpoint_hash.clone()) - }, - code_fraction: if event.code_fraction > 0.0 { - Some(f64::from(event.code_fraction)) - } else { - None - }, - tags, - session_key_hash: if event.session_key_hash.is_empty() { - None - } else { - Some(event.session_key_hash.clone()) - }, - is_prefix_repeat: if event.is_prefix_repeat { - Some(true) - } else { - None - }, - is_code_context_repeat: if event.is_code_context_repeat { - Some(true) - } else { - None - }, - novel_token_count: if event.novel_token_count > 0 { - Some(u64::from(event.novel_token_count)) - } else { - None - }, - repeated_token_count: if event.repeated_token_count > 0 { - Some(u64::from(event.repeated_token_count)) - } else { - None - }, - first_step_event_id: event.first_step_event_id.clone(), - original_event_id: event.original_event_id.clone(), - data_source: enum_name(&event.data_source), - dynamic_fraction: if event.dynamic_fraction > 0.0 { - Some(event.dynamic_fraction) - } else { - None - }, - system_prompt_hash: event.system_prompt_hash.clone(), - tool_definition_hash: event.tool_definition_hash.clone(), - prefix_repeat_signature: event.prefix_repeat_signature.clone(), - complexity_score: if event.complexity_score > 0 { - Some(event.complexity_score) - } else { - None - }, - actual_output_tokens: event.actual_output_tokens, - finish_reason: event.finish_reason.clone(), - response_latency_ms: event.response_latency_ms, - ttfb_ms: event.ttfb_ms, - session_request_count: event.session_request_count, - session_total_tokens: event.session_total_tokens, - session_credential_alerts: event.session_credential_alerts, - conversation_turn: event.conversation_turn, - ws_turn_number: event.ws_turn_number, - session_id: event.session_id.map(|u| u.to_string()), - product_id: event.product_id.clone(), - surface_type: enum_name(&event.surface_type), - is_shadow_it: if event.is_shadow_it { Some(true) } else { None }, - interaction_mode: enum_name(&event.interaction_mode), - } -} - -fn enum_name(value: &T) -> Option { - match serde_json::to_value(value).ok()? { - serde_json::Value::String(value) => Some(value), - _ => None, - } -} - -fn enum_names(values: &[T]) -> Vec { - values.iter().filter_map(enum_name).collect() -} - -fn true_option(value: bool) -> Option { - if value { - Some(true) - } else { - None - } -} - fn normalize_device_id_hash(raw: String) -> String { let trimmed = raw.trim(); if trimmed.is_empty() {