diff --git a/harness/Cargo.lock b/harness/Cargo.lock index 279d67e45..452332fc2 100644 --- a/harness/Cargo.lock +++ b/harness/Cargo.lock @@ -502,13 +502,13 @@ dependencies = [ [[package]] name = "harness" -version = "1.0.3" +version = "1.0.4" dependencies = [ "anyhow", "async-trait", "clap", "globset", - "iii-observability", + "iii-helpers", "iii-sdk", "jsonschema", "schemars", @@ -763,16 +763,18 @@ dependencies = [ ] [[package]] -name = "iii-observability" -version = "0.19.4" +name = "iii-helpers" +version = "0.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71a3002534a53d85c86e4be167cf7ae8c1de809ab3130ecfcb2d85eb4afd1272" +checksum = "09daa7c14a9e4c1f7077c4a181918d207e3f05cdfea8b2d7781bbeb80caf4c5d" dependencies = [ "futures-util", "opentelemetry", "opentelemetry-http", "opentelemetry_sdk", "reqwest", + "schemars", + "serde", "serde_json", "sysinfo", "tokio", @@ -783,14 +785,14 @@ dependencies = [ [[package]] name = "iii-sdk" -version = "0.19.2" +version = "0.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8490f2ad470d54cf7e0bc2f4105aca71bdb4c24752fcb406183f24b8dd328f80" +checksum = "5957568413b9a5178c11bf91b20909e93e568b187e259e941ba6009a7ec3f5c1" dependencies = [ "async-trait", "futures-util", "hostname", - "iii-observability", + "iii-helpers", "reqwest", "schemars", "serde", diff --git a/harness/Cargo.toml b/harness/Cargo.toml index 29a428c8e..69dc483e4 100644 --- a/harness/Cargo.toml +++ b/harness/Cargo.toml @@ -15,10 +15,10 @@ name = "harness" path = "src/lib.rs" [dependencies] -iii-sdk = "=0.19.2" +iii-sdk = "=0.20.0" # Re-exports the same `opentelemetry` the SDK uses, so we can carry the active # OTel context across `tokio::spawn` boundaries (shared `Context` global). -iii-observability = "0.19.2" +iii-helpers = "=0.20.0" tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/harness/src/clients/context.rs b/harness/src/clients/context.rs index aeca5a8de..024deabf5 100644 --- a/harness/src/clients/context.rs +++ b/harness/src/clients/context.rs @@ -4,7 +4,8 @@ use std::sync::Arc; -use iii_sdk::{TriggerRequest, III}; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::IIIClient; use serde::Deserialize; use serde_json::{json, Value}; @@ -44,12 +45,12 @@ pub struct AssembleParams { #[derive(Clone)] pub struct ContextClient { - iii: Arc, + iii: Arc, timeout_ms: u64, } impl ContextClient { - pub fn new(iii: Arc, timeout_ms: u64) -> Self { + pub fn new(iii: Arc, timeout_ms: u64) -> Self { Self { iii, timeout_ms } } diff --git a/harness/src/clients/engine.rs b/harness/src/clients/engine.rs index d0227c31a..95f2932ef 100644 --- a/harness/src/clients/engine.rs +++ b/harness/src/clients/engine.rs @@ -5,7 +5,8 @@ use std::sync::Arc; -use iii_sdk::{TriggerRequest, III}; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::IIIClient; use serde_json::{json, Value}; use crate::error::HarnessError; @@ -21,12 +22,12 @@ pub struct FunctionDescriptor { #[derive(Clone)] pub struct EngineClient { - iii: Arc, + iii: Arc, timeout_ms: u64, } impl EngineClient { - pub fn new(iii: Arc, timeout_ms: u64) -> Self { + pub fn new(iii: Arc, timeout_ms: u64) -> Self { Self { iii, timeout_ms } } diff --git a/harness/src/clients/router.rs b/harness/src/clients/router.rs index e1dff7645..ff10c5c28 100644 --- a/harness/src/clients/router.rs +++ b/harness/src/clients/router.rs @@ -13,9 +13,10 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use async_trait::async_trait; -use iii_observability::opentelemetry::trace::FutureExt as _; +use iii_helpers::observability::opentelemetry::trace::FutureExt as _; use iii_sdk::helpers::create_channel; -use iii_sdk::{TriggerRequest, III}; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::IIIClient; use serde_json::{json, Value}; use tokio::sync::mpsc; use tokio::sync::Notify; @@ -53,13 +54,13 @@ pub struct ChatOutcome { #[derive(Clone)] pub struct RouterClient { - iii: Arc, + iii: Arc, timeout_ms: u64, coalesce_ms: u64, } impl RouterClient { - pub fn new(iii: Arc, timeout_ms: u64, coalesce_ms: u64) -> Self { + pub fn new(iii: Arc, timeout_ms: u64, coalesce_ms: u64) -> Self { Self { iii, timeout_ms, @@ -141,7 +142,7 @@ impl RouterClient { // the SDK uses to attach a handler's context. let iii = self.iii.clone(); let timeout_ms = self.timeout_ms; - let parent_cx = iii_observability::opentelemetry::Context::current(); + let parent_cx = iii_helpers::observability::opentelemetry::Context::current(); let trigger = tokio::spawn( async move { iii.trigger(TriggerRequest { diff --git a/harness/src/clients/session.rs b/harness/src/clients/session.rs index 8097c49c9..eb87c4bc9 100644 --- a/harness/src/clients/session.rs +++ b/harness/src/clients/session.rs @@ -3,7 +3,8 @@ use std::sync::Arc; -use iii_sdk::{TriggerRequest, III}; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::IIIClient; use serde::Deserialize; use serde_json::{json, Value}; @@ -30,12 +31,12 @@ pub struct LoadedCustom { #[derive(Clone)] pub struct SessionClient { - iii: Arc, + iii: Arc, timeout_ms: u64, } impl SessionClient { - pub fn new(iii: Arc, timeout_ms: u64) -> Self { + pub fn new(iii: Arc, timeout_ms: u64) -> Self { Self { iii, timeout_ms } } diff --git a/harness/src/configuration.rs b/harness/src/configuration.rs index cc3c697cc..eee742307 100644 --- a/harness/src/configuration.rs +++ b/harness/src/configuration.rs @@ -19,7 +19,10 @@ use std::sync::Arc; use std::time::Duration; -use iii_sdk::{IIIError, RegisterFunction, RegisterTriggerInput, Trigger, TriggerRequest, III}; +use iii_sdk::errors::Error; +use iii_sdk::protocol::{RegisterTriggerInput, TriggerRequest}; +use iii_sdk::trigger::Trigger; +use iii_sdk::{IIIClient, RegisterFunction}; use serde_json::{json, Value}; use tokio::sync::RwLock; @@ -38,7 +41,7 @@ const CONFIG_RETRY_BACKOFF_MS: u64 = 250; /// Register the `harness` configuration schema. When `seed` is present its /// value is installed as `initial_value`; otherwise the built-in default is /// seeded only when nothing is stored yet (safe to call every boot). -pub async fn register_config(iii: &III, seed: Option<&WorkerConfig>) -> Result<(), String> { +pub async fn register_config(iii: &IIIClient, seed: Option<&WorkerConfig>) -> Result<(), String> { discard_legacy_harness_config_if_needed(iii).await?; let mut payload = json!({ @@ -61,7 +64,7 @@ pub async fn register_config(iii: &III, seed: Option<&WorkerConfig>) -> Result<( /// Read the live `harness` configuration (env-expanded by the configuration /// worker — `from_json` does NOT re-expand). -pub async fn fetch_config(iii: &III) -> Result { +pub async fn fetch_config(iii: &IIIClient) -> Result { let value = get_config_value(iii).await?; if value.is_null() { tracing::info!("no configuration value found; using built-in default configuration"); @@ -70,7 +73,7 @@ pub async fn fetch_config(iii: &III) -> Result { WorkerConfig::from_json(&value) } -async fn should_seed_default_value(iii: &III) -> Result { +async fn should_seed_default_value(iii: &IIIClient) -> Result { match try_get_config_value(iii).await? { None => Ok(true), Some(value) if value.is_null() => Ok(true), @@ -78,14 +81,14 @@ async fn should_seed_default_value(iii: &III) -> Result { } } -async fn get_config_value(iii: &III) -> Result { +async fn get_config_value(iii: &IIIClient) -> Result { try_get_config_value(iii) .await? .ok_or_else(|| format!("configuration `{CONFIG_ID}` not found")) } /// Returns `Ok(None)` when the entry does not exist (codes vary in case). -async fn try_get_config_value(iii: &III) -> Result, String> { +async fn try_get_config_value(iii: &IIIClient) -> Result, String> { match trigger_with_retry(iii, "configuration::get", json!({ "id": CONFIG_ID })).await { Ok(resp) => Ok(resp.get("value").cloned()), Err(e) if e.to_ascii_uppercase().contains("NOT_FOUND") => Ok(None), @@ -121,7 +124,7 @@ fn legacy_harness_keys(value: &Value) -> Vec<&'static str> { /// can succeed. There is no `configuration::delete`; unlock a permissive schema, /// replace the value with built-in defaults, then let the caller register the /// real schema. -async fn discard_legacy_harness_config_if_needed(iii: &III) -> Result<(), String> { +async fn discard_legacy_harness_config_if_needed(iii: &IIIClient) -> Result<(), String> { let Some(value) = try_get_config_value(iii).await? else { return Ok(()); }; @@ -178,7 +181,7 @@ pub struct TriggerHandles { /// Best-effort binding: the cron trigger type always exists (engine built-in), /// but a transient failure must not brick boot — it surfaces as a `None` /// handle. -fn bind(iii: &III, trigger_type: &str, function_id: &str, config: Value) -> Option { +fn bind(iii: &IIIClient, trigger_type: &str, function_id: &str, config: Value) -> Option { match iii.register_trigger(RegisterTriggerInput { trigger_type: trigger_type.to_string(), function_id: function_id.to_string(), @@ -197,7 +200,7 @@ fn bind(iii: &III, trigger_type: &str, function_id: &str, config: Value) -> Opti } /// (Re)bind the cron pending-sweep from the current config. -pub fn bind_sweep(iii: &III, cfg: &WorkerConfig) -> Option { +pub fn bind_sweep(iii: &IIIClient, cfg: &WorkerConfig) -> Option { bind( iii, "cron", @@ -238,10 +241,10 @@ pub struct OnConfigChangeResponse { /// trigger. `handles` holds the live cron `Trigger` the handler re-binds when /// `sweep_expression` changes. pub fn register_config_trigger( - iii: &Arc, + iii: &Arc, cell: ConfigCell, handles: Arc, -) -> Result<(), IIIError> { +) -> Result<(), Error> { let cell_for_fn = cell.clone(); let handles_for_fn = handles.clone(); let engine = iii.clone(); @@ -253,7 +256,7 @@ pub fn register_config_trigger( let engine = engine.clone(); async move { on_config_change(&engine, &cell, &handles).await; - Ok::(OnConfigChangeResponse { ok: true }) + Ok::(OnConfigChangeResponse { ok: true }) } }) .description( @@ -277,7 +280,7 @@ pub fn register_config_trigger( /// Reload from the AUTHORITATIVE configuration. The caller-supplied trigger /// payload is intentionally ignored: a direct call can never inject config. -async fn on_config_change(iii: &III, cell: &ConfigCell, handles: &TriggerHandles) { +async fn on_config_change(iii: &IIIClient, cell: &ConfigCell, handles: &TriggerHandles) { let cfg = match fetch_config(iii).await { Ok(cfg) => cfg, Err(e) => { @@ -296,7 +299,11 @@ async fn on_config_change(iii: &III, cell: &ConfigCell, handles: &TriggerHandles tracing::info!("harness configuration reloaded"); } -async fn trigger_with_retry(iii: &III, function_id: &str, payload: Value) -> Result { +async fn trigger_with_retry( + iii: &IIIClient, + function_id: &str, + payload: Value, +) -> Result { let mut last_err = String::new(); for attempt in 1..=CONFIG_RETRIES { match iii diff --git a/harness/src/deps.rs b/harness/src/deps.rs index c3a3ca58e..91f8ff9db 100644 --- a/harness/src/deps.rs +++ b/harness/src/deps.rs @@ -4,7 +4,7 @@ use std::sync::Arc; -use iii_sdk::III; +use iii_sdk::IIIClient; use crate::clients::{ ContextClient, EngineClient, FunctionDescriptor, RouterClient, SessionClient, @@ -18,7 +18,7 @@ use crate::locks::SessionLocks; #[derive(Clone)] pub struct Deps { - pub iii: Arc, + pub iii: Arc, pub config: ConfigCell, pub functions: FunctionsCell, pub events: TurnEvents, @@ -28,7 +28,7 @@ pub struct Deps { impl Deps { pub fn new( - iii: Arc, + iii: Arc, config: ConfigCell, functions: FunctionsCell, events: TurnEvents, diff --git a/harness/src/discovery.rs b/harness/src/discovery.rs index 58f7b3607..82d624df2 100644 --- a/harness/src/discovery.rs +++ b/harness/src/discovery.rs @@ -17,7 +17,9 @@ use std::sync::Arc; -use iii_sdk::{IIIError, RegisterFunction, RegisterTriggerInput, III}; +use iii_sdk::errors::Error; +use iii_sdk::protocol::RegisterTriggerInput; +use iii_sdk::{IIIClient, RegisterFunction}; use serde_json::json; use tokio::sync::RwLock; @@ -40,7 +42,7 @@ pub async fn apply(cell: &FunctionsCell, functions: Vec) { } /// Fetch the authoritative registry and swap the snapshot; returns the count. -async fn reload(iii: &Arc, cell: &FunctionsCell, timeout_ms: u64) -> usize { +async fn reload(iii: &Arc, cell: &FunctionsCell, timeout_ms: u64) -> usize { let engine = EngineClient::new(iii.clone(), timeout_ms); let functions = engine.functions_list().await; let count = functions.len(); @@ -50,7 +52,7 @@ async fn reload(iii: &Arc, cell: &FunctionsCell, timeout_ms: u64) -> usize /// Seed the snapshot from the registry. The trigger fires only on change, so /// without this the cache would stay empty until the first change. -pub async fn seed(iii: &Arc, cell: &FunctionsCell, timeout_ms: u64) { +pub async fn seed(iii: &Arc, cell: &FunctionsCell, timeout_ms: u64) { let count = reload(iii, cell, timeout_ms).await; tracing::info!(count, "seeded function-registry cache"); } @@ -75,7 +77,7 @@ pub struct OnFunctionsChangeResponse { /// seeded snapshot still serves — it just won't update until restart) rather /// than bricking boot. The handler is tagged `internal` so it stays off the /// public catalog (and out of the very cache it maintains). -pub fn register_functions_trigger(iii: &Arc, cell: FunctionsCell, timeout_ms: u64) { +pub fn register_functions_trigger(iii: &Arc, cell: FunctionsCell, timeout_ms: u64) { let engine = iii.clone(); iii.register_function( FUNCTIONS_FN_ID, @@ -85,7 +87,7 @@ pub fn register_functions_trigger(iii: &Arc, cell: FunctionsCell, timeout_m async move { let count = reload(&engine, &cell, timeout_ms).await; tracing::debug!(count, "function-registry cache refreshed"); - Ok::(OnFunctionsChangeResponse { ok: true }) + Ok::(OnFunctionsChangeResponse { ok: true }) } }) .description( diff --git a/harness/src/error.rs b/harness/src/error.rs index 1cecae369..07461b387 100644 --- a/harness/src/error.rs +++ b/harness/src/error.rs @@ -4,7 +4,7 @@ //! `harness/spawn_depth_exceeded`, and `harness/spawn_fanout_exceeded` //! verbatim. -use iii_sdk::IIIError; +use iii_sdk::errors::Error; #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] pub enum HarnessError { @@ -52,9 +52,9 @@ impl HarnessError { } } -impl From for IIIError { +impl From for Error { fn from(e: HarnessError) -> Self { - IIIError::Handler(e.to_string()) + Error::Handler(e.to_string()) } } diff --git a/harness/src/events.rs b/harness/src/events.rs index 0b966b6d8..55dba73d8 100644 --- a/harness/src/events.rs +++ b/harness/src/events.rs @@ -12,10 +12,10 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; use async_trait::async_trait; -use iii_sdk::{ - IIIError, RegisterTriggerType, TriggerAction, TriggerConfig, TriggerHandler, TriggerRequest, - III, -}; +use iii_sdk::errors::Error; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::trigger::{TriggerConfig, TriggerHandler}; +use iii_sdk::{IIIClient, RegisterTriggerType, TriggerAction}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -118,15 +118,15 @@ struct TurnEventTriggerHandler { #[async_trait] impl TriggerHandler for TurnEventTriggerHandler { - async fn register_trigger(&self, config: TriggerConfig) -> Result<(), IIIError> { + async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> { let id = config.id.clone(); let function_id = config.function_id.clone(); - self.set.add(config).map_err(IIIError::Handler)?; + self.set.add(config).map_err(Error::Handler)?; tracing::info!(trigger_type = self.type_id, %id, %function_id, "turn-event subscription registered"); Ok(()) } - async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), IIIError> { + async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> { self.set.remove(&config.id); Ok(()) } @@ -136,7 +136,7 @@ impl TriggerHandler for TurnEventTriggerHandler { /// fan-out. Cloned into [`crate::deps::Deps`]. #[derive(Clone)] pub struct TurnEvents { - iii: Arc, + iii: Arc, started: SubscriberSet, completed: SubscriberSet, } @@ -144,7 +144,7 @@ pub struct TurnEvents { impl TurnEvents { /// Register both trigger types and return the emitter. Must run before /// function registration so the handlers capture the subscriber sets. - pub fn register(iii: &Arc) -> Self { + pub fn register(iii: &Arc) -> Self { let started = SubscriberSet::default(); let completed = SubscriberSet::default(); diff --git a/harness/src/functions/mod.rs b/harness/src/functions/mod.rs index 39878941b..1f71171a5 100644 --- a/harness/src/functions/mod.rs +++ b/harness/src/functions/mod.rs @@ -15,7 +15,8 @@ pub mod turn; use std::future::Future; use std::sync::Arc; -use iii_sdk::{IIIError, RegisterFunction, III}; +use iii_sdk::errors::Error; +use iii_sdk::{IIIClient, RegisterFunction}; use schemars::JsonSchema; use serde::de::DeserializeOwned; use serde::Serialize; @@ -55,7 +56,7 @@ pub const STATUS_DESC: &str = "Read the current turn status for a session."; /// Register one typed handler under `id`, mapping `HarnessError` into the bus /// error shape (`code: message`). fn register( - iii: &Arc, + iii: &Arc, deps: &Arc, id: &str, description: &str, @@ -72,13 +73,13 @@ fn register( RegisterFunction::new_async(move |req: Req| { let deps = deps.clone(); let handler = handler.clone(); - async move { handler(deps, req).await.map_err(IIIError::from) } + async move { handler(deps, req).await.map_err(Error::from) } }) .description(description), ); } -pub fn register_all(iii: &Arc, deps: &Arc) { +pub fn register_all(iii: &Arc, deps: &Arc) { register(iii, deps, SEND_ID, SEND_DESC, |d, r| async move { send::handle(&d, r).await }); diff --git a/harness/src/functions/turn.rs b/harness/src/functions/turn.rs index d985d311c..faf4d1593 100644 --- a/harness/src/functions/turn.rs +++ b/harness/src/functions/turn.rs @@ -20,7 +20,7 @@ pub async fn handle(deps: &Deps, payload: TurnStepPayload) -> Result Result { diff --git a/harness/src/hooks/mod.rs b/harness/src/hooks/mod.rs index b44300c68..7212439fd 100644 --- a/harness/src/hooks/mod.rs +++ b/harness/src/hooks/mod.rs @@ -9,7 +9,9 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; use async_trait::async_trait; -use iii_sdk::{IIIError, RegisterTriggerType, TriggerConfig, TriggerHandler, III}; +use iii_sdk::errors::Error; +use iii_sdk::trigger::{TriggerConfig, TriggerHandler}; +use iii_sdk::{IIIClient, RegisterTriggerType}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -169,17 +171,15 @@ struct HookTriggerHandler { #[async_trait] impl TriggerHandler for HookTriggerHandler { - async fn register_trigger(&self, config: TriggerConfig) -> Result<(), IIIError> { + async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> { let id = config.id.clone(); let function_id = config.function_id.clone(); - self.set - .add(self.point, config) - .map_err(IIIError::Handler)?; + self.set.add(self.point, config).map_err(Error::Handler)?; tracing::info!(trigger_type = self.point.trigger_type(), %id, %function_id, "hook binding registered"); Ok(()) } - async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), IIIError> { + async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> { self.set.remove(&config.id); Ok(()) } @@ -189,7 +189,7 @@ impl TriggerHandler for HookTriggerHandler { /// invocation. Cloned into [`crate::deps::Deps`]. #[derive(Clone)] pub struct HookRegistry { - pub iii: Arc, + pub iii: Arc, pub pre_turn: HookSet, pub pre_generate: HookSet, pub post_generate: HookSet, @@ -200,7 +200,7 @@ pub struct HookRegistry { impl HookRegistry { /// Register the five hook trigger types and return the registry. Must run /// before function registration so handlers capture the sets. - pub fn register(iii: &Arc) -> Self { + pub fn register(iii: &Arc) -> Self { let registry = HookRegistry { iii: iii.clone(), pre_turn: HookSet::default(), @@ -218,7 +218,7 @@ impl HookRegistry { registry } - fn register_type(&self, iii: &Arc, point: HookPoint, set: HookSet) { + fn register_type(&self, iii: &Arc, point: HookPoint, set: HookSet) { let description = match point { HookPoint::PreTurn => "Synchronous hook: first step of a turn, before any model spend. May veto.", HookPoint::PreGenerate => "Synchronous hook: after context assembly, before generation. May extend the system prompt, append messages, or veto.", diff --git a/harness/src/hooks/runner.rs b/harness/src/hooks/runner.rs index d5565e1ef..205046b27 100644 --- a/harness/src/hooks/runner.rs +++ b/harness/src/hooks/runner.rs @@ -7,7 +7,7 @@ use std::time::Duration; use globset::{Glob, GlobSetBuilder}; -use iii_sdk::TriggerRequest; +use iii_sdk::protocol::TriggerRequest; use serde_json::{Map, Value}; use super::{HookBinding, HookPoint, HookRegistry}; diff --git a/harness/src/main.rs b/harness/src/main.rs index 6f6bf4f34..224a2c121 100644 --- a/harness/src/main.rs +++ b/harness/src/main.rs @@ -21,7 +21,8 @@ use std::sync::Arc; use anyhow::{Context, Result}; use clap::Parser; -use iii_sdk::{register_worker, InitOptions, WorkerMetadata}; +use iii_sdk::runtime::WorkerMetadata; +use iii_sdk::{register_worker, InitOptions}; use tokio::sync::RwLock; use harness::configuration::{self, ConfigCell, TriggerHandles}; diff --git a/harness/src/state.rs b/harness/src/state.rs index 86713caf5..4644aaa97 100644 --- a/harness/src/state.rs +++ b/harness/src/state.rs @@ -6,7 +6,8 @@ //! returns the stored value directly (null when absent); `state::delete` //! returns the prior value. -use iii_sdk::{TriggerRequest, III}; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::IIIClient; use serde_json::{json, Value}; use crate::error::HarnessError; @@ -16,7 +17,7 @@ pub const TURN_SCOPE: &str = "harness_turn"; pub const IDEM_SCOPE: &str = "harness_idem"; async fn state_get( - iii: &III, + iii: &IIIClient, scope: &str, key: &str, timeout_ms: u64, @@ -32,7 +33,7 @@ async fn state_get( } async fn state_set( - iii: &III, + iii: &IIIClient, scope: &str, key: &str, value: Value, @@ -50,7 +51,7 @@ async fn state_set( } async fn state_delete( - iii: &III, + iii: &IIIClient, scope: &str, key: &str, timeout_ms: u64, @@ -68,7 +69,7 @@ async fn state_delete( /// Read the turn record for a session (`None` when absent or null). pub async fn get_turn( - iii: &III, + iii: &IIIClient, session_id: &str, timeout_ms: u64, ) -> Result, HarnessError> { @@ -83,19 +84,27 @@ pub async fn get_turn( /// Persist the turn record (whole-record write; the loop holds the only /// writer per session via the per-session lock). -pub async fn put_turn(iii: &III, record: &TurnRecord, timeout_ms: u64) -> Result<(), HarnessError> { +pub async fn put_turn( + iii: &IIIClient, + record: &TurnRecord, + timeout_ms: u64, +) -> Result<(), HarnessError> { let value = serde_json::to_value(record) .map_err(|e| HarnessError::State(format!("turn record serialize: {e}")))?; state_set(iii, TURN_SCOPE, &record.session_id, value, timeout_ms).await } -pub async fn delete_turn(iii: &III, session_id: &str, timeout_ms: u64) -> Result<(), HarnessError> { +pub async fn delete_turn( + iii: &IIIClient, + session_id: &str, + timeout_ms: u64, +) -> Result<(), HarnessError> { state_delete(iii, TURN_SCOPE, session_id, timeout_ms).await } /// List every turn record (the pending-call sweep scans these). `state::list` /// returns a values array (or an object map); both shapes are tolerated. -pub async fn list_turns(iii: &III, timeout_ms: u64) -> Result, HarnessError> { +pub async fn list_turns(iii: &IIIClient, timeout_ms: u64) -> Result, HarnessError> { let v = iii .trigger(TriggerRequest { function_id: "state::list".into(), @@ -129,7 +138,7 @@ fn parse_record_list(v: &Value) -> Vec { } pub async fn get_idem( - iii: &III, + iii: &IIIClient, key: &str, timeout_ms: u64, ) -> Result, HarnessError> { @@ -143,7 +152,7 @@ pub async fn get_idem( } pub async fn put_idem( - iii: &III, + iii: &IIIClient, key: &str, record: &IdemRecord, timeout_ms: u64, diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs index fd41f85e0..339ebbe2d 100644 --- a/harness/src/turn_loop.rs +++ b/harness/src/turn_loop.rs @@ -5,7 +5,8 @@ //! deterministic entry ids, and per-call checkpoints make redelivery safe. use async_trait::async_trait; -use iii_sdk::{TriggerAction, TriggerRequest, III}; +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::{IIIClient, TriggerAction}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; @@ -43,7 +44,7 @@ pub struct TurnStepResult { /// Enqueue the next durable loop step onto the engine's `default` queue. pub async fn enqueue_step( - iii: &III, + iii: &IIIClient, session_id: &str, turn_id: &str, step: u64,