zeph-core 0.19.3

Core agent loop, configuration, context builder, metrics, and vault for Zeph
Documentation
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
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! Streaming partial-JSON parser for speculative tool-call dispatch.
//!
//! The Anthropic SSE tool-use stream emits a single JSON object across many
//! `input_json_delta` events. Standard JSON parsers (including
//! `serde_json::StreamDeserializer`) wait for the closing brace before yielding
//! any value — providing no benefit for speculative dispatch.
//!
//! `PartialJsonParser` is a ~120-line brace/string/escape state machine that
//! accumulates delta strings and extracts top-level leaf keys whose values have
//! been **fully closed** (primitives, fully closed nested objects/arrays).
//! When all required fields of a tool's `input_schema` are present, the engine
//! can speculatively dispatch the tool call without waiting for `ToolUseStop`.
//!
//! ## Invariants
//!
//! - **Escape state**: the escape flag is set after a literal `\` inside a string
//!   and cleared after the following character, regardless of what that character is.
//!   Multi-byte escape sequences (e.g. `\uXXXX`) are not individually validated;
//!   the parser only tracks structural JSON tokens.
//! - **Mid-array edge case**: array values at depth > 1 are treated as opaque; their
//!   contents are never surfaced as `known_leaves`. Only top-level keys (depth == 1)
//!   whose values close cleanly are included.
//! - **No allocation on malformed input**: `Malformed` is returned immediately when
//!   a structural invariant is violated; the buffer is not reallocated.

#![allow(dead_code)]

use serde_json::Map;

/// Result of feeding accumulated JSON delta bytes to [`PartialJsonParser::push`].
#[derive(Debug, Clone, PartialEq)]
pub enum PrefixState {
    /// Input is still inside an unterminated string, imbalanced braces, or ends mid-escape.
    Incomplete,
    /// A valid prefix has been parsed: the `known_leaves` map contains fully closed top-level
    /// key-value pairs. `missing_required` lists required schema keys not yet seen.
    ValidPrefix {
        /// Top-level keys whose values are fully closed.
        known_leaves: Map<String, serde_json::Value>,
        /// Required tool input schema keys not yet present in the buffer.
        missing_required: Vec<String>,
    },
    /// The buffer contains a sequence that cannot be a valid JSON object prefix.
    Malformed,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Ctx {
    TopObject,
    InKey,
    AfterKey,
    InValue,
    InStringValue,
    InNestedValue { depth: u32 },
}

/// Streaming structural parser for partial Anthropic SSE tool-input JSON.
///
/// Feed each `InputJsonDelta` string via [`push`](Self::push). When the result is
/// [`PrefixState::ValidPrefix`] with an empty `missing_required`, the engine may
/// synthesize a `ToolCall` and speculatively dispatch it.
///
/// # Examples
///
/// ```rust
/// use zeph_core::agent::speculative::partial_json::{PartialJsonParser, PrefixState};
///
/// let mut p = PartialJsonParser::new();
/// // Simulate incremental SSE deltas
/// p.push(r#"{"command": "ls "#);
/// let state = p.push(r#"-la"}"#);
/// assert!(matches!(state, PrefixState::ValidPrefix { .. }));
/// ```
pub struct PartialJsonParser {
    buf: String,
    required: Vec<String>,
    /// Cached known leaves from the last successful scan.
    known_cache: Map<String, serde_json::Value>,
    /// Byte offset in `buf` up to which the cache is valid.
    /// When `buf.len() > scan_watermark` we re-scan only the tail, rebuilding from the
    /// cached state (H2: avoids O(N²) full-buffer rescan on every push).
    scan_watermark: usize,
}

impl PartialJsonParser {
    /// Create a new parser. Call [`set_required`](Self::set_required) to configure
    /// the list of required schema keys before the first [`push`](Self::push).
    #[must_use]
    pub fn new() -> Self {
        Self {
            buf: String::new(),
            required: Vec::new(),
            known_cache: Map::new(),
            scan_watermark: 0,
        }
    }

    /// Set the list of keys required by the tool's `input_schema`.
    ///
    /// Keys in this list that are absent from the accumulated buffer are reported as
    /// `missing_required` in [`PrefixState::ValidPrefix`].
    pub fn set_required(&mut self, required: Vec<String>) {
        self.required = required;
    }

    /// Append `delta` and re-scan the buffer.
    ///
    /// Returns the current [`PrefixState`]. May be called repeatedly; each call
    /// replaces the previous result.
    pub fn push(&mut self, delta: &str) -> PrefixState {
        self.buf.push_str(delta);
        self.scan()
    }

    /// Reset the parser for reuse after a commit or cancel.
    pub fn reset(&mut self) {
        self.buf.clear();
        self.known_cache.clear();
        self.scan_watermark = 0;
    }

    /// Scan `self.buf` and extract fully closed top-level key-value pairs.
    ///
    /// Uses a watermark cursor to avoid replaying already-scanned bytes (H2).
    /// Previously confirmed key-value pairs are carried in `known_cache`; only the
    /// bytes after `scan_watermark` are newly examined.
    fn scan(&mut self) -> PrefixState {
        let bytes = self.buf.as_bytes();
        let len = bytes.len();

        // On the first call, consume the opening '{'.
        let mut i = if self.scan_watermark == 0 {
            let start = skip_ws(bytes, 0);
            if start >= len || bytes[start] != b'{' {
                return if self.buf.trim().is_empty() {
                    PrefixState::Incomplete
                } else {
                    PrefixState::Malformed
                };
            }
            start + 1 // consume '{'
        } else {
            self.scan_watermark
        };

        // Start with previously confirmed pairs; we may add more this call.
        let mut known = self.known_cache.clone();

        loop {
            i = skip_ws(bytes, i);
            if i >= len {
                break; // still incomplete
            }

            // End of object
            if bytes[i] == b'}' {
                self.scan_watermark = i + 1;
                let missing = self.missing(&known);
                self.known_cache.clone_from(&known);
                return PrefixState::ValidPrefix {
                    known_leaves: known,
                    missing_required: missing,
                };
            }

            // Comma between pairs
            if bytes[i] == b',' {
                i += 1;
                i = skip_ws(bytes, i);
                if i >= len {
                    break;
                }
            }

            // Key string
            if bytes[i] != b'"' {
                return PrefixState::Malformed;
            }
            let Some((key, after_key)) = read_string(bytes, i) else {
                break; // incomplete string
            };
            i = after_key;
            i = skip_ws(bytes, i);

            if i >= len {
                break;
            }
            if bytes[i] != b':' {
                return PrefixState::Malformed;
            }
            i += 1; // consume ':'
            i = skip_ws(bytes, i);

            if i >= len {
                break; // value not yet arrived
            }

            // Try to read a fully closed value
            match read_value(bytes, i) {
                ReadValue::Complete(value, end) => {
                    known.insert(key, value);
                    // Advance watermark: this pair is confirmed; next call starts here.
                    self.scan_watermark = end;
                    self.known_cache.clone_from(&known);
                    i = end;
                }
                ReadValue::Incomplete => break,
                ReadValue::Malformed => return PrefixState::Malformed,
            }
        }

        // Buffer ended mid-object — partial but valid so far
        let missing = self.missing(&known);
        PrefixState::ValidPrefix {
            known_leaves: known,
            missing_required: missing,
        }
    }

    fn missing(&self, known: &Map<String, serde_json::Value>) -> Vec<String> {
        self.required
            .iter()
            .filter(|k| !known.contains_key(k.as_str()))
            .cloned()
            .collect()
    }
}

impl Default for PartialJsonParser {
    fn default() -> Self {
        Self::new()
    }
}

// --- Internal helpers -------------------------------------------------------

fn skip_ws(bytes: &[u8], mut i: usize) -> usize {
    while i < bytes.len() && matches!(bytes[i], b' ' | b'\t' | b'\r' | b'\n') {
        i += 1;
    }
    i
}

/// Read a JSON string starting at `i` (which must point at `"`).
/// Returns `(string_content, index_after_closing_quote)` or `None` if incomplete.
///
/// Operates on raw bytes to avoid the Latin-1 `b as char` cast that corrupts non-ASCII.
/// The closed string slice is validated as UTF-8 and decoded at object boundary.
fn read_string(bytes: &[u8], start: usize) -> Option<(String, usize)> {
    debug_assert_eq!(bytes[start], b'"');
    let mut i = start + 1;
    let mut escape = false;
    while i < bytes.len() {
        let b = bytes[i];
        if escape {
            escape = false;
        } else if b == b'\\' {
            escape = true;
        } else if b == b'"' {
            // Decode the full slice [start+1..i] as UTF-8 in one shot.
            let content = std::str::from_utf8(&bytes[start + 1..i]).ok()?;
            // Re-process escape sequences by delegating to serde_json, which handles
            // \uXXXX, \n, \t, etc. correctly without re-implementing the JSON string spec.
            let json_str = [b"\"", &bytes[start + 1..i], b"\""].concat();
            let decoded: String =
                serde_json::from_slice(&json_str).unwrap_or_else(|_| content.to_owned());
            return Some((decoded, i + 1));
        }
        i += 1;
    }
    None // unterminated string
}

enum ReadValue {
    Complete(serde_json::Value, usize),
    Incomplete,
    Malformed,
}

/// Attempt to read a fully closed JSON value starting at `i`.
fn read_value(bytes: &[u8], i: usize) -> ReadValue {
    if i >= bytes.len() {
        return ReadValue::Incomplete;
    }
    match bytes[i] {
        b'"' => match read_string(bytes, i) {
            Some((s, end)) => ReadValue::Complete(serde_json::Value::String(s), end),
            None => ReadValue::Incomplete,
        },
        b'{' | b'[' => read_nested(bytes, i),
        b't' => read_literal(bytes, i, b"true", serde_json::Value::Bool(true)),
        b'f' => read_literal(bytes, i, b"false", serde_json::Value::Bool(false)),
        b'n' => read_literal(bytes, i, b"null", serde_json::Value::Null),
        b'-' | b'0'..=b'9' => read_number(bytes, i),
        _ => ReadValue::Malformed,
    }
}

fn read_literal(bytes: &[u8], i: usize, lit: &[u8], val: serde_json::Value) -> ReadValue {
    if bytes.len() < i + lit.len() {
        return ReadValue::Incomplete;
    }
    if &bytes[i..i + lit.len()] == lit {
        ReadValue::Complete(val, i + lit.len())
    } else {
        ReadValue::Malformed
    }
}

fn read_number(bytes: &[u8], mut i: usize) -> ReadValue {
    let start = i;
    if i < bytes.len() && bytes[i] == b'-' {
        i += 1;
    }
    while i < bytes.len() && (bytes[i].is_ascii_digit() || bytes[i] == b'.') {
        i += 1;
    }
    // Exponent
    if i < bytes.len() && matches!(bytes[i], b'e' | b'E') {
        i += 1;
        if i < bytes.len() && matches!(bytes[i], b'+' | b'-') {
            i += 1;
        }
        while i < bytes.len() && bytes[i].is_ascii_digit() {
            i += 1;
        }
    }
    if i == start {
        return ReadValue::Malformed;
    }
    // Must be followed by structural char or end
    if i < bytes.len() && !matches!(bytes[i], b',' | b'}' | b']' | b' ' | b'\t' | b'\r' | b'\n') {
        return ReadValue::Incomplete;
    }
    let s = std::str::from_utf8(&bytes[start..i]).unwrap_or("");
    match serde_json::from_str::<serde_json::Value>(s) {
        Ok(v) => ReadValue::Complete(v, i),
        Err(_) => ReadValue::Malformed,
    }
}

/// Read a nested `{...}` or `[...]`, tracking depth and string escapes.
fn read_nested(bytes: &[u8], start: usize) -> ReadValue {
    let open = bytes[start];
    let close = if open == b'{' { b'}' } else { b']' };
    let mut depth = 1u32;
    let mut i = start + 1;
    let mut in_string = false;
    let mut escape = false;

    while i < bytes.len() {
        let b = bytes[i];
        if escape {
            escape = false;
        } else if in_string {
            if b == b'\\' {
                escape = true;
            } else if b == b'"' {
                in_string = false;
            }
        } else if b == b'"' {
            in_string = true;
        } else if b == open {
            depth += 1;
        } else if b == close {
            depth -= 1;
            if depth == 0 {
                // Re-parse just the slice [start..=i] as JSON
                let parsed = std::str::from_utf8(&bytes[start..=i])
                    .ok()
                    .and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok());
                return match parsed {
                    Some(v) => ReadValue::Complete(v, i + 1),
                    None => ReadValue::Malformed,
                };
            }
        }
        i += 1;
    }
    ReadValue::Incomplete
}

// ---------------------------------------------------------------------------
// Unit tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    fn push_all(p: &mut PartialJsonParser, parts: &[&str]) -> PrefixState {
        let mut state = PrefixState::Incomplete;
        for part in parts {
            state = p.push(part);
        }
        state
    }

    /// Fixture 1: simple single-field bash command arriving in two deltas.
    #[test]
    fn fixture_simple_command_two_deltas() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["command".into()]);
        p.push(r#"{"command": "ls "#);
        let state = p.push(r#"-la"}"#);
        match state {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(missing_required.is_empty());
                let v = known_leaves["command"].as_str().unwrap();
                assert!(v.contains("ls") && v.contains("la"), "got: {v}");
            }
            other => panic!("expected ValidPrefix, got {other:?}"),
        }
    }

    /// Fixture 2: multi-field tool call with nested object arrives incrementally.
    #[test]
    fn fixture_multi_field_incremental() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["path".into(), "content".into()]);
        let state = push_all(
            &mut p,
            &[
                r#"{"path": "/tmp/f"#,
                r#"oo.txt", "content": "hel"#,
                r#"lo world"}"#,
            ],
        );
        match state {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(missing_required.is_empty(), "missing: {missing_required:?}");
                assert!(known_leaves.contains_key("path"));
                assert!(known_leaves.contains_key("content"));
            }
            other => panic!("expected ValidPrefix, got {other:?}"),
        }
    }

    /// Fixture 3: escape sequence inside string value does not break parser.
    #[test]
    fn fixture_escape_in_string() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["msg".into()]);
        let state = p.push(r#"{"msg": "say \"hello\""}"#);
        match state {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(missing_required.is_empty());
                let v = known_leaves["msg"].as_str().unwrap();
                assert!(v.contains("hello"), "got: {v}");
            }
            other => panic!("expected ValidPrefix, got {other:?}"),
        }
    }

    /// Fixture 4: mid-delta truncation → Incomplete, then resolved on next delta.
    #[test]
    fn fixture_incomplete_then_resolved() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["x".into()]);
        let mid = p.push(r#"{"x": 42"#);
        // No closing brace yet; key present but object not closed → ValidPrefix with x known
        match &mid {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(missing_required.is_empty());
                assert_eq!(known_leaves["x"], 42);
            }
            PrefixState::Incomplete => {} // also acceptable
            other @ PrefixState::Malformed => panic!("unexpected: {other:?}"),
        }
        let done = p.push("}");
        assert!(matches!(done, PrefixState::ValidPrefix { .. }));
    }

    /// Fixture 5: malformed input returns Malformed.
    #[test]
    fn fixture_malformed_input() {
        let mut p = PartialJsonParser::new();
        let state = p.push("not-json");
        assert!(matches!(state, PrefixState::Malformed));
    }

    /// Fixture 6: mid-array value at top level is opaque (depth > 1 skipped).
    #[test]
    fn fixture_top_level_array_value() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["items".into()]);
        let state = p.push(r#"{"items": [1, 2, 3]}"#);
        match state {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(missing_required.is_empty());
                assert!(known_leaves["items"].is_array());
            }
            other => panic!("expected ValidPrefix, got {other:?}"),
        }
    }

    #[test]
    fn reset_clears_buffer() {
        let mut p = PartialJsonParser::new();
        p.push(r#"{"x": 1}"#);
        p.reset();
        let state = p.push(r#"{"y": 2}"#);
        match state {
            PrefixState::ValidPrefix { known_leaves, .. } => {
                assert!(
                    !known_leaves.contains_key("x"),
                    "should be cleared after reset"
                );
            }
            other => panic!("{other:?}"),
        }
    }

    /// Fixture 7: Unicode / non-ASCII bytes must not be corrupted (C5 regression).
    #[test]
    fn fixture_unicode_filename() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["path".into()]);
        let state = p.push(r#"{"path": "/tmp/Привет.txt"}"#);
        match state {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(missing_required.is_empty());
                let v = known_leaves["path"].as_str().unwrap();
                assert!(v.contains("Привет"), "non-ASCII corrupted: {v}");
            }
            other => panic!("expected ValidPrefix, got {other:?}"),
        }
    }

    /// Fixture 8: incremental watermark — second push does not re-parse completed pairs (H2).
    #[test]
    fn fixture_incremental_watermark() {
        let mut p = PartialJsonParser::new();
        p.set_required(vec!["a".into(), "b".into()]);
        // First push: 'a' complete, 'b' incomplete
        let s1 = p.push(r#"{"a": 1, "b": "#);
        match &s1 {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(known_leaves.contains_key("a"));
                assert!(missing_required.contains(&"b".to_string()));
            }
            PrefixState::Incomplete => {} // also acceptable before 'b' is fully parsed
            other @ PrefixState::Malformed => panic!("unexpected s1: {other:?}"),
        }
        // Second push: completes 'b'
        let s2 = p.push("2}");
        match s2 {
            PrefixState::ValidPrefix {
                known_leaves,
                missing_required,
            } => {
                assert!(
                    missing_required.is_empty(),
                    "still missing: {missing_required:?}"
                );
                assert_eq!(known_leaves["a"], 1);
                assert_eq!(known_leaves["b"], 2);
            }
            other => panic!("expected ValidPrefix, got {other:?}"),
        }
    }
}