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
|
//! Pull-based streaming surface for `libpanto`.
//!
//! This is the spine of the language-bindings work (see
//! `docs/libpanto-bindings.md` and `docs/phase0-pull-stream-design.md`).
//! Instead of pushing events at a `Receiver` vtable, the agent loop is
//! inverted into a resumable `Stream` whose `next()` *pulls* one `Event`
//! at a time. Pull is the more primitive primitive: push composes trivially
//! on top of it, and it maps 1:1 onto Go range-over-func iterators and
//! Python generators.
//!
//! Contract (the terminal-event invariant):
//!
//! - `Event` (a value) -> streaming progress, including `turn_complete`.
//! - `null` -> the stream is exhausted (already past the
//! terminal `turn_complete`). Never returned
//! before `turn_complete`.
//! - `error.X` -> a genuine failure (network, parse, provider).
//!
//! Event payloads borrow from state owned by the stream or the
//! conversation. **An `Event` is valid only until the next `next()` call.**
//! Consumers that need to retain data copy it out before advancing.
const std = @import("std");
const Allocator = std.mem.Allocator;
const conversation = @import("conversation.zig");
const provider_mod = @import("provider.zig");
const session_mod = @import("session.zig");
pub const ContentBlockType = provider_mod.ContentBlockType;
pub const Usage = session_mod.Usage;
pub const ProviderRetryInfo = provider_mod.ProviderRetryInfo;
/// The single success-only event type every binding marshals. Mirrors the
/// former `ReceiverVTable` callbacks plus the agent's tool-dispatch
/// boundaries. Provider failures are NOT a variant here — they surface as
/// the `!` in `Stream.next() !?Event`.
pub const Event = union(enum) {
/// An assistant message began streaming.
message_start: conversation.MessageRole,
/// A content block opened. `index` is the receiver-facing block index.
block_start: BlockStart,
/// Tool identity resolved for a ToolUse block (id + name both known).
/// Fires once per ToolUse block, after its `block_start` and before its
/// `block_complete`. Borrowed slices; valid until the next `next()`.
tool_details: ToolDetails,
/// Streaming content for the open block (text, thinking, or tool args).
/// `delta` is borrowed; valid until the next `next()`.
content_delta: ContentDelta,
/// A content block closed. `block` is borrowed from the message under
/// construction; valid until the next `next()`.
block_complete: BlockComplete,
/// One assistant message finished streaming (one provider response).
/// In a tool-using turn this fires once per assistant message, not once
/// per turn — `turn_complete` is the turn-level terminal. `message` is
/// borrowed from the conversation.
message_complete: MessageComplete,
/// Provider retry scheduled before the agent sleeps for the next
/// attempt. Purely informational; simple consumers ignore it.
provider_retry: ProviderRetryInfo,
/// The agent began dispatching the tool calls in the just-completed
/// assistant message. Marks the boundary between a provider stream and
/// concurrent tool execution.
tool_dispatch_start: ToolDispatchStart,
/// The agent finished dispatching tools and appended a user(ToolResult)
/// message to the conversation. `message` is borrowed.
tool_dispatch_complete: ToolDispatchComplete,
/// The turn terminal: the model stopped calling tools and the turn is
/// done. Emitted exactly once, after the final `message_complete` and
/// any tool dispatch. Every `next()` after this returns `null`.
turn_complete,
pub const BlockStart = struct {
block_type: ContentBlockType,
index: usize,
};
pub const ToolDetails = struct {
index: usize,
id: []const u8,
name: []const u8,
};
pub const ContentDelta = struct {
index: usize,
delta: []const u8,
};
pub const BlockComplete = struct {
index: usize,
block: conversation.ContentBlock,
};
pub const MessageComplete = struct {
message: conversation.Message,
usage: ?Usage,
};
pub const ToolDispatchStart = struct {
count: usize,
};
pub const ToolDispatchComplete = struct {
message: conversation.Message,
};
};
/// A small FIFO of decoded-but-not-yet-yielded events. One `parser.feed()`
/// can yield several SSE events, each of which can produce several `Event`s;
/// the provider decode step appends them here and `Stream.next()` drains
/// the queue before pulling more bytes.
///
/// Transient byte payloads (delta text, tool id/name) are duped into a
/// queue-owned arena via `dupeBytes`, so they survive the provider freeing
/// its SSE/JSON scratch and any reallocation of the provider's accumulation
/// buffers. The arena (and the event list) are reset when the queue fully
/// drains, so memory is bounded by the events produced from a single byte
/// chunk. Events whose payloads live in the conversation (`block_complete`,
/// `message_complete`, `tool_dispatch_complete`) borrow directly and are
/// not duped — the conversation outlives the `next()` step.
pub const EventQueue = struct {
items: std.ArrayList(Event) = .empty,
head: usize = 0,
arena: std.heap.ArenaAllocator,
allocator: Allocator,
pub fn init(allocator: Allocator) EventQueue {
return .{
.allocator = allocator,
.arena = std.heap.ArenaAllocator.init(allocator),
};
}
pub fn deinit(self: *EventQueue) void {
self.items.deinit(self.allocator);
self.arena.deinit();
}
pub fn push(self: *EventQueue, ev: Event) !void {
try self.items.append(self.allocator, ev);
}
/// Copy transient bytes into the queue's arena. The returned slice is
/// valid until the queue next fully drains (i.e. until the consumer has
/// pulled every queued event). Providers MUST route any byte payload
/// borrowed from SSE/JSON scratch or a reallocating buffer through here
/// before queueing it on an event.
pub fn dupeBytes(self: *EventQueue, bytes: []const u8) ![]const u8 {
return self.arena.allocator().dupe(u8, bytes);
}
pub fn isEmpty(self: *const EventQueue) bool {
return self.head >= self.items.items.len;
}
/// Pop the next event, or null if empty. Resets the backing list and
/// arena when drained so they can be refilled for the next chunk without
/// unbounded growth.
pub fn pop(self: *EventQueue) ?Event {
if (self.head >= self.items.items.len) {
// Drained: reset list + arena to reuse for the next chunk.
self.items.clearRetainingCapacity();
self.head = 0;
_ = self.arena.reset(.retain_capacity);
return null;
}
const ev = self.items.items[self.head];
self.head += 1;
return ev;
}
};
|
