1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
use std::collections::BTreeMap;
use std::fmt::Display;
use std::sync::{Arc, OnceLock, RwLock};
use serde_json::{Map, Value};
use crate::AppError;
use crate::workflow::definition::JsonBoundary;
use crate::workflow::{StateSerialization, WorkflowState};
#[derive(Clone, Default)]
pub(super) struct SerializationRegistry {
fields: BTreeMap<String, FieldCodec>,
tags: BTreeMap<String, Arc<dyn ValueCodec>>,
}
#[derive(Clone)]
struct FieldCodec {
serialize: Arc<dyn ValueCodec>,
deserialize: Arc<dyn ValueCodec>,
}
trait ValueCodec: Send + Sync {
fn transform(&self, value: &Value) -> Result<Value, AppError>;
}
impl<F> ValueCodec for F
where
F: Fn(&Value) -> Result<Value, AppError> + Send + Sync,
{
fn transform(&self, value: &Value) -> Result<Value, AppError> {
self(value)
}
}
impl StateSerialization {
/// Returns the one process-global registry, shared across all state owners.
///
/// ```
/// use app_kit::workflow::{StateSerialization, WorkflowState};
/// assert!(std::ptr::eq(StateSerialization::global(), StateSerialization::global()));
/// StateSerialization::global().register_tag("global_counter".into(), |value| {
/// value.get("count").cloned().ok_or("missing count")
/// })?;
/// let encoded = serde_json::from_str(r#"{"count":{"__burr_serde__":"global_counter","count":3}}"#)?;
/// assert_eq!(WorkflowState::deserialize(encoded)?.values()["count"], 3);
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
pub fn global() -> &'static Self {
static REGISTRY: OnceLock<StateSerialization> = OnceLock::new();
REGISTRY.get_or_init(|| Self {
registry: RwLock::new(Arc::new(SerializationRegistry::default())),
})
}
/// Registers or replaces both transformations for one exact top-level field.
///
/// Encoders must return objects; decoders may return any valid JSON value.
/// Callback errors retain their Display text plus field and direction context.
/// Neither callback invocation nor destruction of replaced captures holds
/// the registry lock. An in-flight operation keeps its original callbacks.
///
/// ```
/// use app_kit::workflow::{StateSerialization, WorkflowState};
/// use serde_json::{json, Map, Value};
/// StateSerialization::global().register_field(
/// "distance_meters".into(),
/// |value| Ok::<_, String>(json!({"meters": value})),
/// |value| value.get("meters").cloned().ok_or("missing meters"),
/// )?;
/// let state = WorkflowState::new(Map::from_iter([
/// ("distance_meters".into(), Value::from(12)),
/// ]))?;
/// let encoded = state.serialize()?;
/// assert_eq!(encoded["distance_meters"], json!({"meters": 12}));
/// assert_eq!(WorkflowState::deserialize(encoded)?, state);
/// # Ok::<(), app_kit::AppError>(())
/// ```
///
/// # Errors
///
/// Returns `Integration` if the registry lock is poisoned. Callback results are
/// validated when encoding/decoding, not during registration; exact field names
/// replace prior registrations for future operations.
pub fn register_field<S, D, SE, DE>(
&self,
field: String,
serialize: S,
deserialize: D,
) -> Result<(), AppError>
where
S: Fn(&Value) -> Result<Value, SE> + Send + Sync + 'static,
D: Fn(&Value) -> Result<Value, DE> + Send + Sync + 'static,
SE: Display,
DE: Display,
{
let encoding_field = field.clone();
let decoding_field = field.clone();
let codec = FieldCodec {
serialize: Arc::new(move |value: &Value| {
serialize(value).map_err(|error| {
AppError::InvalidInput(format!(
"field serialize handler for '{encoding_field}' failed: {error}"
))
})
}),
deserialize: Arc::new(move |value: &Value| {
deserialize(value).map_err(|error| {
AppError::InvalidInput(format!(
"field deserialize handler for '{decoding_field}' failed: {error}"
))
})
}),
};
let replaced = {
let mut registry = self.registry.write().map_err(|_| {
AppError::Integration("state serialization registry lock was poisoned".to_owned())
})?;
Arc::make_mut(&mut registry).fields.insert(field, codec)
};
drop(replaced);
Ok(())
}
/// Registers or replaces a decoder for an exact `__burr_serde__` tag string.
///
/// Empty strings are valid names. The decoder receives the complete original
/// tagged object. Its output is terminal: it is not dispatched or recursively
/// decoded again. Ordinary untagged containers are decoded recursively.
///
/// ```
/// use app_kit::workflow::{StateSerialization, WorkflowState};
/// use serde_json::{json, Map};
/// StateSerialization::global().register_tag("distance".into(), |value| {
/// value.get("meters").cloned().ok_or("missing meters")
/// })?;
/// let state = WorkflowState::deserialize(Map::from_iter([(
/// "route".into(),
/// json!([{"__burr_serde__": "distance", "meters": 12}]),
/// )]))?;
/// assert_eq!(state.values()["route"], json!([12]));
/// # Ok::<(), app_kit::AppError>(())
/// ```
///
/// # Errors
///
/// Returns `Integration` if the registry lock is poisoned. Decoder failures occur
/// at decode time and preserve caller error text with tag context.
pub fn register_tag<D, DE>(&self, tag: String, deserialize: D) -> Result<(), AppError>
where
D: Fn(&Value) -> Result<Value, DE> + Send + Sync + 'static,
DE: Display,
{
let decoding_tag = tag.clone();
let decoder = Arc::new(move |value: &Value| {
deserialize(value).map_err(|error| {
AppError::InvalidInput(format!(
"serde dispatch handler for '{decoding_tag}' failed: {error}"
))
})
});
let replaced = {
let mut registry = self.registry.write().map_err(|_| {
AppError::Integration("state serialization registry lock was poisoned".to_owned())
})?;
Arc::make_mut(&mut registry).tags.insert(tag, decoder)
};
drop(replaced);
Ok(())
}
fn snapshot(&self) -> Result<Arc<SerializationRegistry>, AppError> {
self.registry
.read()
.map(|registry| Arc::clone(®istry))
.map_err(|_| {
AppError::Integration("state serialization registry lock was poisoned".to_owned())
})
}
}
impl WorkflowState {
/// Encodes ordered state using one snapshot of the global field registry.
///
/// Unregistered fields keep their JSON values, including tagged objects.
/// Field encoders must produce objects. The complete encoded state retains
/// the ordinary JSON byte/depth bounds; failure leaves this state unchanged.
///
/// ```
/// use app_kit::workflow::WorkflowState;
/// use serde_json::{Map, Value};
/// let values = Map::from_iter([("count".into(), Value::from(2))]);
/// let state = WorkflowState::new(values.clone())?;
/// assert_eq!(state.serialize()?, values);
/// # Ok::<(), app_kit::AppError>(())
/// ```
///
/// # Errors
///
/// Returns `InvalidInput` for a codec error, a field encoder returning a nonobject,
/// or exceeded JSON bounds; `Integration` reports registry lock poisoning.
///
/// # Panics
///
/// Panics if a registered field encoder panics. The registry lock is released
/// before callbacks run.
pub fn serialize(&self) -> Result<Map<String, Value>, AppError> {
StateSerialization::global()
.snapshot()?
.encode(self.values())
}
/// Decodes state with one snapshot of global field and tag registrations.
///
/// Top-level field decoders take precedence over recursive tagged decoding.
/// Missing and malformed tags fail; handler outputs are not redispatched.
/// Returned state preserves key order and passes the JSON boundary checks.
///
/// ```
/// use app_kit::workflow::WorkflowState;
/// let values = serde_json::from_str(r#"{"count":2,"ready":true}"#)?;
/// let state = WorkflowState::deserialize(values)?;
/// assert_eq!(state.values()["count"], 2);
/// assert_eq!(state.values()["ready"], true);
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
///
/// # Errors
///
/// Returns `InvalidInput` for malformed state, a nonstring or unregistered tag,
/// a decoder error, or invalid transformed values. Registry lock poisoning is
/// `Integration`; field/tag decoder outputs are not recursively redispatched.
///
/// # Panics
///
/// Panics if a registered field or tag decoder panics. The registry lock is
/// released before callbacks run.
pub fn deserialize(values: Map<String, Value>) -> Result<Self, AppError> {
JsonBoundary::validate_map(&values)?;
let values = StateSerialization::global()
.snapshot()?
.decode_state(values)?;
Self::new(values)
}
}
impl SerializationRegistry {
fn encode(&self, state: &Map<String, Value>) -> Result<Map<String, Value>, AppError> {
let mut output = Map::with_capacity(state.len());
for (field, value) in state {
let value = match self.fields.get(field) {
Some(codec) => {
let value = codec.serialize.transform(value)?;
if !value.is_object() {
return Err(AppError::InvalidInput(format!(
"field serialize handler for '{field}' must return an object, got {}",
Self::value_kind(&value)
)));
}
value
}
None => value.clone(),
};
output.insert(field.clone(), value);
}
JsonBoundary::validate_map(&output)?;
Ok(output)
}
fn decode_state(&self, state: Map<String, Value>) -> Result<Map<String, Value>, AppError> {
let mut output = Map::with_capacity(state.len());
for (field, value) in state {
let value = match self.fields.get(&field) {
Some(codec) => codec.deserialize.transform(&value)?,
None => self.decode_value(value)?,
};
output.insert(field, value);
}
Ok(output)
}
fn decode_value(&self, value: Value) -> Result<Value, AppError> {
match value {
Value::Array(values) => values
.into_iter()
.map(|value| self.decode_value(value))
.collect::<Result<Vec<_>, _>>()
.map(Value::Array),
Value::Object(values) => {
if let Some(tag) = values.get("__burr_serde__") {
let name = tag.as_str().ok_or_else(|| {
AppError::InvalidInput(format!(
"dispatch key '__burr_serde__' must be a string, got {}",
Self::value_kind(tag)
))
})?;
let decoder = self.tags.get(name).ok_or_else(|| {
AppError::InvalidInput(format!("no function registered for key: {name}"))
})?;
return decoder.transform(&Value::Object(values));
}
let mut output = Map::with_capacity(values.len());
for (key, value) in values {
output.insert(key, self.decode_value(value)?);
}
Ok(Value::Object(output))
}
value => Ok(value),
}
}
fn value_kind(value: &Value) -> &'static str {
match value {
Value::Null => "null",
Value::Bool(_) => "bool",
Value::Number(number) if number.is_i64() => "i64",
Value::Number(number) if number.is_u64() => "u64",
Value::Number(_) => "f64",
Value::String(_) => "string",
Value::Array(_) => "array",
Value::Object(_) => "object",
}
}
}