libpanto fixes

- ensuring stream.next() sequences are interruptible
- Agent.run() takes any user message, not just chat text

10 files changed, 649 insertions, 285 deletions

diff --git a/examples/simple-agent-go/main.go b/examples/simple-agent-go/main.go
index c733cd7..b8e11d6 100644
--- a/examples/simple-agent-go/main.go
+++ b/examples/simple-agent-go/main.go
@@ -101,20 +101,20 @@ func main() {
 		},
 	})
 
-	stream, err := agent.Run(prompt)
+	stream, err := agent.Run(panto.UserMessage(panto.TextBlock(prompt)))
 	if err != nil {
 		panic(err)
 	}
 	defer stream.Close()
 
 	for ev := range stream.Iter() {
-		switch ev.Tag {
-		case panto.EventContentDelta:
-			fmt.Print(ev.ContentDelta.Delta)
+		switch ev := ev.(type) {
+		case panto.ContentDeltaEvent:
+			fmt.Print(ev.Delta)
 			os.Stdout.Sync()
-		case panto.EventToolDispatchStart:
-			fmt.Fprintf(os.Stderr, "\n[bash: running %d tool call(s)]\n", ev.ToolDispatchStart.Count)
-		case panto.EventToolDispatchComplete:
+		case panto.ToolDispatchStartEvent:
+			fmt.Fprintf(os.Stderr, "\n[bash: running %d tool call(s)]\n", ev.Count)
+		case panto.ToolDispatchCompleteEvent:
 			fmt.Fprint(os.Stderr, "[bash: done]\n")
 		}
 	}
diff --git a/libpanto-c/include/panto.h b/libpanto-c/include/panto.h
index 3b4a22a..c79e77c 100644
--- a/libpanto-c/include/panto.h
+++ b/libpanto-c/include/panto.h
@@ -50,7 +50,11 @@ typedef struct { PantoContentBlockTag block_type; size_t index; } PantoBlockStar
 typedef struct { size_t index; PantoSlice id, name; } PantoToolDetails;
 typedef struct { size_t index; PantoSlice delta; } PantoContentDelta;
 typedef struct { size_t index; PantoContentBlockTag block_type; } PantoBlockComplete;
-typedef struct { bool has_usage; PantoUsage usage; } PantoMessageComplete;
+/* One assistant message finished streaming. `message_index` indexes the
+ * agent's live conversation (`panto_agent_conversation`), whose tail is this
+ * message when the event is delivered; fetch its `block_count` blocks from
+ * there for full content. `usage` is the wire-reported token usage. */
+typedef struct { PantoMessageRole role; size_t message_index; size_t block_count; bool has_usage; PantoUsage usage; } PantoMessageComplete;
 typedef struct { bool compacted; size_t kept_turns; size_t summarized_messages; } PantoCompactionResult;
 typedef struct { size_t attempt, max_attempts; uint64_t delay_ms; PantoSlice error_name; bool has_status_code; uint16_t status_code; bool has_retry_after_ms; uint64_t retry_after_ms; PantoSlice message; bool compaction; } PantoProviderRetry;
 typedef struct { size_t count; } PantoToolDispatchStart;
@@ -181,7 +185,12 @@ PANTO_EXPORT PantoStatus panto_agent_set_config(PantoAgent *agent, const PantoCo
 PANTO_EXPORT PantoStatus panto_agent_add_system_message(PantoAgent *agent, const char *text);
 PANTO_EXPORT PantoStatus panto_agent_set_system_prompt(PantoAgent *agent, const char *text);
 PANTO_EXPORT PantoStatus panto_agent_compact(PantoAgent *agent, const char *override_system_prompt, const char *extra_instructions, PantoCompactionResult *out);
-PANTO_EXPORT PantoStatus panto_agent_run(PantoAgent *agent, const char *user_text, PantoStream **out);
+/* Open a turn from a user-role message built with the PantoMessageBuilder API
+ * (the same builder used to rebuild persisted messages): one `add_text` block
+ * for a plain chat turn, or `add_tool_result` blocks to resume after the
+ * embedder handled tool calls itself. Consumes `user_message` (the builder is
+ * destroyed); do not reuse or destroy it afterwards. */
+PANTO_EXPORT PantoStatus panto_agent_run(PantoAgent *agent, PantoMessageBuilder *user_message, PantoStream **out);
 /* Reset a failed stream back to its turn-open boundary so the caller can
  * resume panto_stream_next() after changing the agent config (e.g. catch a
  * terminal bad-request, swap the provider config, and retry the SAME turn
diff --git a/libpanto-c/src/lib.zig b/libpanto-c/src/lib.zig
index 9d584a0..7f61ae3 100644
--- a/libpanto-c/src/lib.zig
+++ b/libpanto-c/src/lib.zig
@@ -50,7 +50,7 @@ pub const PantoBlockStart = extern struct { block_type: PantoContentBlockTag, in
 pub const PantoToolDetails = extern struct { index: usize, id: PantoSlice, name: PantoSlice };
 pub const PantoContentDelta = extern struct { index: usize, delta: PantoSlice };
 pub const PantoBlockComplete = extern struct { index: usize, block_type: PantoContentBlockTag };
-pub const PantoMessageComplete = extern struct { has_usage: bool, usage: PantoUsage };
+pub const PantoMessageComplete = extern struct { role: PantoMessageRole, message_index: usize, block_count: usize, has_usage: bool, usage: PantoUsage };
 pub const PantoCompactionResult = extern struct { compacted: bool, kept_turns: usize, summarized_messages: usize };
 pub const PantoProviderRetry = extern struct { attempt: usize, max_attempts: usize, delay_ms: u64, error_name: PantoSlice, has_status_code: bool, status_code: u16, has_retry_after_ms: bool, retry_after_ms: u64, message: PantoSlice, compaction: bool };
 pub const PantoToolDispatchStart = extern struct { count: usize };
@@ -803,9 +803,21 @@ export fn panto_agent_compact(a: *PantoAgent, override_system_prompt: ?[*:0]cons
     out.* = .{ .compacted = r.compacted, .kept_turns = r.kept_turns, .summarized_messages = r.summarized_messages };
     return .ok;
 }
-export fn panto_agent_run(a: *PantoAgent, text: ?[*:0]const u8, out: **PantoStream) PantoStatus {
+/// Open a turn from a user-role message built with the `PantoMessageBuilder`
+/// API — the same builder used to rebuild persisted messages. A plain chat
+/// turn is one `add_text` block; a turn that resumes after the embedder handled
+/// tool calls is one or more `add_tool_result` blocks. Consumes the builder
+/// (success or failure): `Agent.run` adopts the block bytes, so only the
+/// builder's list backing + shell are freed here, never the per-block bytes.
+export fn panto_agent_run(a: *PantoAgent, user_message: *PantoMessageBuilder, out: **PantoStream) PantoStatus {
     const h: *AgentHandle = @ptrCast(@alignCast(a));
-    const st = h.agent.run(.{ .text = cstr(text) }) catch |e| return setErr("{t}", .{e});
+    const b = unwrapBuilder(user_message);
+    defer {
+        b.blocks.deinit(allocator);
+        if (b.identity) |id| panto.freeWireIdentity(allocator, id);
+        allocator.destroy(b);
+    }
+    const st = h.agent.run(.{ .blocks = b.blocks.items }) catch |e| return setErr("{t}", .{e});
     out.* = @ptrCast(st);
     return .ok;
 }
@@ -825,6 +837,14 @@ export fn panto_stream_next(s: *PantoStream, out: *PantoEvent) PantoNextStatus {
     const ev = st.next() catch |e| return setErrNext("{t}", .{e});
     if (ev) |e| {
         out.* = marshalEvent(e) catch |err| return setErrNext("{t}", .{err});
+        // The just-completed assistant message is the conversation's current
+        // tail (no further message is committed within this `next()` call), so
+        // its index is `len - 1`. Stamp it so the embedder can fetch the full
+        // message via `panto_agent_conversation` rather than marshalling its
+        // blocks onto every event.
+        if (out.tag == .message_complete) {
+            out.data.message_complete.message_index = st._agent.conversation.messages.items.len - 1;
+        }
         return .event;
     }
     return .done;
@@ -855,7 +875,7 @@ fn marshalEvent(ev: panto.Event) !PantoEvent {
         .tool_details => |t| .{ .tag = .tool_details, .data = .{ .tool_details = .{ .index = t.index, .id = try dupOut(t.id), .name = try dupOut(t.name) } } },
         .content_delta => |d| .{ .tag = .content_delta, .data = .{ .content_delta = .{ .index = d.index, .delta = try dupOut(d.delta) } } },
         .block_complete => |b| .{ .tag = .block_complete, .data = .{ .block_complete = .{ .index = b.index, .block_type = @enumFromInt(@intFromEnum(b.block)) } } },
-        .message_complete => |m| .{ .tag = .message_complete, .data = .{ .message_complete = .{ .has_usage = m.usage != null, .usage = if (m.usage) |u| usage(u) else .{ .input = 0, .output = 0, .cache_read = 0, .cache_write = 0, .reasoning = 0 } } } },
+        .message_complete => |m| .{ .tag = .message_complete, .data = .{ .message_complete = .{ .role = @enumFromInt(@intFromEnum(m.message.role)), .message_index = 0, .block_count = m.message.content.items.len, .has_usage = m.usage != null, .usage = if (m.usage) |u| usage(u) else .{ .input = 0, .output = 0, .cache_read = 0, .cache_write = 0, .reasoning = 0 } } } },
         .provider_retry => |r| .{ .tag = .provider_retry, .data = .{ .provider_retry = .{ .attempt = r.attempt, .max_attempts = r.max_attempts, .delay_ms = r.delay_ms, .error_name = try dupOut(@errorName(r.err)), .has_status_code = r.status_code != null, .status_code = r.status_code orelse 0, .has_retry_after_ms = r.retry_after_ms != null, .retry_after_ms = r.retry_after_ms orelse 0, .message = try dupOut(r.message orelse ""), .compaction = r.compaction } } },
         .tool_dispatch_start => |t| .{ .tag = .tool_dispatch_start, .data = .{ .tool_dispatch_start = .{ .count = t.count } } },
         .tool_dispatch_result => .{ .tag = .tool_dispatch_result, .data = .{ .tool_dispatch_result = .{ .message_index = 0 } } },
diff --git a/libpanto-go/README.md b/libpanto-go/README.md
index a19e788..063f800 100644
--- a/libpanto-go/README.md
+++ b/libpanto-go/README.md
@@ -37,12 +37,20 @@ agent, err := panto.NewAgent(panto.Config{
 if err != nil { /* handle */ }
 defer agent.Close()
 
-stream, err := agent.Run("hello")
+stream, err := agent.Run(panto.UserMessage(panto.TextBlock("hello")))
 if err != nil { /* handle */ }
 defer stream.Close()
 
 for ev := range stream.Iter() {
-    // ev.Tag identifies the active payload.
+    // ev is an Event interface; type-switch on the concrete event, or branch
+    // on ev.EventType(). Breaking out of the loop is safe: the session store
+    // is current at every step.
+    switch ev := ev.(type) {
+    case panto.ContentDeltaEvent:
+        fmt.Print(ev.Delta)
+    case panto.MessageCompleteEvent:
+        // full message at agent.Conversation().Message(ev.MessageIndex)
+    }
 }
 if err := stream.Err(); err != nil { /* handle */ }
 ```
diff --git a/libpanto-go/panto.go b/libpanto-go/panto.go
index 5bfc5d8..f692b6f 100644
--- a/libpanto-go/panto.go
+++ b/libpanto-go/panto.go
@@ -155,51 +155,88 @@ type (
 	}
 )
 
-type Event struct {
-	Tag                  EventTag
-	MessageStart         MessageRole
-	BlockStart           BlockStart
-	ToolDetails          ToolDetails
-	ContentDelta         ContentDelta
-	BlockComplete        BlockComplete
-	MessageComplete      MessageComplete
-	ProviderRetry        ProviderRetry
-	ToolDispatchStart    ToolDispatchStart
-	ToolDispatchComplete ToolDispatchComplete
+// Event is one streaming-progress notification from a Stream. The concrete
+// type is one of the *Event structs below; type-switch on it, or branch on
+// EventType() when only the kind matters. The terminal event of a turn is
+// TurnCompleteEvent (EventType() == EventTurnComplete).
+type Event interface {
+	// EventType reports the kind of event, mirroring the concrete type. It
+	// lets callers branch without a type switch (e.g. skip everything until
+	// EventMessageComplete) and is the cheap terminal check used by Iter.
+	EventType() EventTag
 }
-type BlockStart struct {
+
+// MessageStartEvent: an assistant message began streaming.
+type MessageStartEvent struct{ Role MessageRole }
+
+// BlockStartEvent: a content block opened.
+type BlockStartEvent struct {
 	BlockType ContentBlockTag
 	Index     uint
 }
-type ToolDetails struct {
+
+// ToolDetailsEvent: the tool identity (id + name) for a ToolUse block resolved.
+type ToolDetailsEvent struct {
 	Index    uint
 	ID, Name string
 }
-type ContentDelta struct {
+
+// ContentDeltaEvent: streaming content for the open block (text/thinking/args).
+type ContentDeltaEvent struct {
 	Index uint
 	Delta string
 }
-type BlockComplete struct {
+
+// BlockCompleteEvent: a content block closed.
+type BlockCompleteEvent struct {
 	Index     uint
 	BlockType ContentBlockTag
 }
-type (
-	MessageComplete struct{ Usage *Usage }
-	ProviderRetry   struct {
-		Attempt, MaxAttempts uint
-		DelayMS              uint64
-		ErrorName            string
-		StatusCode           *uint16
-		RetryAfterMS         *uint64
-		Message              string
-		Compaction           bool
-	}
-)
 
-type (
-	ToolDispatchStart    struct{ Count uint }
-	ToolDispatchComplete struct{ MessageIndex uint }
-)
+// MessageCompleteEvent: one assistant message finished streaming. The message
+// is the current tail of Agent.Conversation(); MessageIndex is its index there
+// and BlockCount its number of content blocks, so the full content can be read
+// back via Agent.Conversation().Message(MessageIndex) without marshalling every
+// block onto the event. Usage is the wire-reported token usage, if any.
+type MessageCompleteEvent struct {
+	Role         MessageRole
+	MessageIndex uint
+	BlockCount   uint
+	Usage        *Usage
+}
+
+// ProviderRetryEvent: a provider retry was scheduled before the next attempt.
+type ProviderRetryEvent struct {
+	Attempt, MaxAttempts uint
+	DelayMS              uint64
+	ErrorName            string
+	StatusCode           *uint16
+	RetryAfterMS         *uint64
+	Message              string
+	Compaction           bool
+}
+
+// ToolDispatchStartEvent: the agent began dispatching the just-completed
+// message's tool calls.
+type ToolDispatchStartEvent struct{ Count uint }
+
+// ToolDispatchCompleteEvent: the agent appended a user(ToolResult) message.
+type ToolDispatchCompleteEvent struct{ MessageIndex uint }
+
+// TurnCompleteEvent: the turn terminal. Emitted exactly once; every Next after
+// it returns nil.
+type TurnCompleteEvent struct{}
+
+func (MessageStartEvent) EventType() EventTag         { return EventMessageStart }
+func (BlockStartEvent) EventType() EventTag           { return EventBlockStart }
+func (ToolDetailsEvent) EventType() EventTag          { return EventToolDetails }
+func (ContentDeltaEvent) EventType() EventTag         { return EventContentDelta }
+func (BlockCompleteEvent) EventType() EventTag        { return EventBlockComplete }
+func (MessageCompleteEvent) EventType() EventTag      { return EventMessageComplete }
+func (ProviderRetryEvent) EventType() EventTag        { return EventProviderRetry }
+func (ToolDispatchStartEvent) EventType() EventTag    { return EventToolDispatchStart }
+func (ToolDispatchCompleteEvent) EventType() EventTag { return EventToolDispatchComplete }
+func (TurnCompleteEvent) EventType() EventTag         { return EventTurnComplete }
 
 type (
 	Agent struct {
@@ -275,8 +312,8 @@ type OwnedContentBlock struct {
 	SystemMode SystemMode
 	// Thinking-block signature provenance. Populated for blocks delivered to
 	// AppendMessages; empty otherwise. To replay a thinking block faithfully,
-	// persist all four and feed them back via MessageBuilder.AddThinking —
-	// Anthropic drops a thinking block whose signature lacks a matching origin.
+	// persist all four and feed them back via ThinkingBlock — Anthropic drops
+	// a thinking block whose signature lacks a matching origin.
 	ThinkingSignature        string
 	ThinkingSignatureStyle   APIStyle
 	ThinkingSignatureBaseURL string
@@ -541,127 +578,198 @@ func (c *Conversation) AddCompactionSummary(text string) error {
 	return nil
 }
 
-// MessageBuilder accumulates content blocks for one message, then commits them
-// to a Conversation via Conversation.AddMessage. It is the block-level builder
-// for rebuilding a persisted conversation that contains tool calls or thinking
-// (the text-only AddUserText/AddAssistantText cannot). Add blocks in order,
-// then call AddMessage (which consumes the builder). Call Destroy only to
-// discard a builder you never commit.
-type MessageBuilder struct {
-	ptr *C.PantoMessageBuilder
+// Block is one immutable content block — text, thinking, tool use, tool
+// result, or system text. Build it with the constructors below (TextBlock,
+// ToolResultBlock, …) and assemble blocks into a message with UserMessage /
+// AssistantMessage. This is the single content-block vocabulary used both to
+// open a turn (Agent.Run) and to rebuild a persisted message for resumption
+// (Conversation.AddMessage).
+type Block struct {
+	tag                ContentBlockTag
+	text               string
+	toolID             string
+	toolName           string
+	input              string
+	isError            bool
+	systemMode         SystemMode
+	thinkingSignature  string
+	thinkingSigStyle   APIStyle
+	thinkingSigBaseURL string
+	thinkingSigModel   string
 }
 
-func NewMessageBuilder(role MessageRole) (*MessageBuilder, error) {
-	p := C.panto_message_builder_create(C.PantoMessageRole(role))
-	if p == nil {
-		return nil, lastError()
-	}
-	b := &MessageBuilder{ptr: p}
-	runtime.SetFinalizer(b, (*MessageBuilder).Destroy)
-	return b, nil
+// TextBlock is a plain text block.
+func TextBlock(text string) Block { return Block{tag: BlockText, text: text} }
+
+// ThinkingBlock is an assistant reasoning block. signature/baseURL/model carry
+// the signature provenance (see OwnedContentBlock); pass "" for signature when
+// the provider emitted none, in which case style/baseURL/model are ignored.
+func ThinkingBlock(text, signature string, style APIStyle, baseURL, model string) Block {
+	return Block{tag: BlockThinking, text: text, thinkingSignature: signature, thinkingSigStyle: style, thinkingSigBaseURL: baseURL, thinkingSigModel: model}
 }
 
-func (b *MessageBuilder) AddText(text string) error {
-	ct := C.CString(text)
-	defer C.free(unsafe.Pointer(ct))
-	if C.panto_message_builder_add_text(b.ptr, ct) != C.PANTO_OK {
-		return lastError()
-	}
-	return nil
+// ToolUseBlock is an assistant tool call.
+func ToolUseBlock(id, name, input string) Block {
+	return Block{tag: BlockToolUse, toolID: id, toolName: name, input: input}
 }
 
-// AddThinking appends a thinking block. signature/baseURL/model carry the
-// signature provenance (see OwnedContentBlock); pass "" for signature when the
-// provider emitted none, in which case style/baseURL/model are ignored.
-func (b *MessageBuilder) AddThinking(text, signature string, style APIStyle, baseURL, model string) error {
-	ct := C.CString(text)
-	defer C.free(unsafe.Pointer(ct))
-	cs := C.CString(signature)
-	defer C.free(unsafe.Pointer(cs))
-	cb := C.CString(baseURL)
-	defer C.free(unsafe.Pointer(cb))
-	cm := C.CString(model)
-	defer C.free(unsafe.Pointer(cm))
-	if C.panto_message_builder_add_thinking(b.ptr, ct, cs, C.PantoAPIStyle(style), cb, cm) != C.PANTO_OK {
-		return lastError()
-	}
-	return nil
+// ToolResultBlock is a user-role tool result keyed by the originating tool-use
+// id. Used to resume a turn whose tool calls the embedder handled itself.
+func ToolResultBlock(toolUseID, text string, isError bool) Block {
+	return Block{tag: BlockToolResult, toolID: toolUseID, text: text, isError: isError}
 }
 
-func (b *MessageBuilder) AddToolUse(id, name, input string) error {
-	cid := C.CString(id)
-	defer C.free(unsafe.Pointer(cid))
-	cn := C.CString(name)
-	defer C.free(unsafe.Pointer(cn))
-	ci := C.CString(input)
-	defer C.free(unsafe.Pointer(ci))
-	if C.panto_message_builder_add_tool_use(b.ptr, cid, cn, ci) != C.PANTO_OK {
-		return lastError()
-	}
-	return nil
+// SystemBlock is a system-prompt block (append or replace mode).
+func SystemBlock(text string, mode SystemMode) Block {
+	return Block{tag: BlockSystem, text: text, systemMode: mode}
 }
 
-func (b *MessageBuilder) AddToolResult(toolUseID, text string, isError bool) error {
-	cid := C.CString(toolUseID)
-	defer C.free(unsafe.Pointer(cid))
-	ct := C.CString(text)
-	defer C.free(unsafe.Pointer(ct))
-	if C.panto_message_builder_add_tool_result(b.ptr, cid, ct, C.bool(isError)) != C.PANTO_OK {
-		return lastError()
-	}
-	return nil
+// MessageIdentity is the wire identity (provider style/baseURL/model +
+// reasoning) that produced a message. Attach it to an InputMessage when
+// rebuilding a conversation for resumption: it is the per-message source of
+// truth for replaying thinking signatures and is preserved through compaction,
+// so a kept-verbatim turn is not re-stamped with the compaction model. Set it
+// from a PersistentMessage's APIStyle/BaseURL/Model/Reasoning.
+type MessageIdentity struct {
+	APIStyle  APIStyle
+	BaseURL   string
+	Model     string
+	Reasoning ReasoningEffort
 }
 
-func (b *MessageBuilder) AddSystem(text string, mode SystemMode) error {
-	ct := C.CString(text)
-	defer C.free(unsafe.Pointer(ct))
-	if C.panto_message_builder_add_system(b.ptr, ct, C.PantoSystemMode(mode)) != C.PANTO_OK {
-		return lastError()
-	}
-	return nil
+// InputMessage is one role-tagged message assembled from value Blocks, with
+// optional usage and wire identity. Build it with UserMessage / AssistantMessage
+// (or set the fields directly), then pass it to Agent.Run (user role) or
+// Conversation.AddMessage (any role, for replay).
+type InputMessage struct {
+	Role     MessageRole
+	Usage    *Usage
+	Identity *MessageIdentity
+	Blocks   []Block
 }
 
-// SetIdentity attaches the wire identity (provider style/baseURL/model +
-// reasoning) that produced this message. This is the per-message source of
-// truth for replaying thinking signatures and is preserved through compaction,
-// so a kept-verbatim turn is not re-stamped with the compaction model. Prefer
-// this over per-block origins on AddThinking when rebuilding a conversation for
-// resumption: set it from PersistentMessage's APIStyle/BaseURL/Model and the
-// thinking signatures replay correctly without copying the identity onto every
-// thinking block.
-func (b *MessageBuilder) SetIdentity(style APIStyle, baseURL, model string, reasoning ReasoningEffort) error {
-	cb := C.CString(baseURL)
-	defer C.free(unsafe.Pointer(cb))
-	cm := C.CString(model)
-	defer C.free(unsafe.Pointer(cm))
-	if C.panto_message_builder_set_identity(b.ptr, C.PantoAPIStyle(style), cb, cm, C.PantoReasoningEffort(reasoning)) != C.PANTO_OK {
-		return lastError()
+// UserMessage assembles a user-role message — the argument to Agent.Run. A
+// plain chat turn is a single TextBlock; a turn resuming after embedder-handled
+// tool calls is one or more ToolResultBlocks.
+func UserMessage(blocks ...Block) InputMessage {
+	return InputMessage{Role: RoleUser, Blocks: blocks}
+}
+
+// AssistantMessage assembles an assistant-role message with optional usage —
+// used to rebuild a persisted turn via Conversation.AddMessage.
+func AssistantMessage(usage *Usage, blocks ...Block) InputMessage {
+	return InputMessage{Role: RoleAssistant, Usage: usage, Blocks: blocks}
+}
+
+// WithIdentity returns a copy of the message tagged with the given wire
+// identity (see MessageIdentity).
+func (m InputMessage) WithIdentity(id MessageIdentity) InputMessage {
+	m.Identity = &id
+	return m
+}
+
+// buildCBuilder materializes an InputMessage as a transient C message builder,
+// pushing each block in order and attaching identity if present. The returned
+// builder is owned by the caller and must be consumed by exactly one of
+// panto_agent_run / panto_conversation_add_message (both of which destroy it).
+// On error it is destroyed here and nil is returned.
+func buildCBuilder(msg InputMessage) (*C.PantoMessageBuilder, error) {
+	p := C.panto_message_builder_create(C.PantoMessageRole(msg.Role))
+	if p == nil {
+		return nil, lastError()
 	}
-	return nil
+	committed := false
+	defer func() {
+		if !committed {
+			C.panto_message_builder_destroy(p)
+		}
+	}()
+	if msg.Identity != nil {
+		cb := C.CString(msg.Identity.BaseURL)
+		defer C.free(unsafe.Pointer(cb))
+		cm := C.CString(msg.Identity.Model)
+		defer C.free(unsafe.Pointer(cm))
+		if C.panto_message_builder_set_identity(p, C.PantoAPIStyle(msg.Identity.APIStyle), cb, cm, C.PantoReasoningEffort(msg.Identity.Reasoning)) != C.PANTO_OK {
+			return nil, lastError()
+		}
+	}
+	for _, b := range msg.Blocks {
+		if err := pushBlock(p, b); err != nil {
+			return nil, err
+		}
+	}
+	committed = true
+	return p, nil
 }
 
-// Destroy frees an uncommitted builder. Safe to call after AddMessage (no-op).
-func (b *MessageBuilder) Destroy() {
-	if b != nil && b.ptr != nil {
-		C.panto_message_builder_destroy(b.ptr)
-		b.ptr = nil
+func pushBlock(p *C.PantoMessageBuilder, b Block) error {
+	switch b.tag {
+	case BlockText:
+		ct := C.CString(b.text)
+		defer C.free(unsafe.Pointer(ct))
+		if C.panto_message_builder_add_text(p, ct) != C.PANTO_OK {
+			return lastError()
+		}
+	case BlockThinking:
+		ct := C.CString(b.text)
+		defer C.free(unsafe.Pointer(ct))
+		cs := C.CString(b.thinkingSignature)
+		defer C.free(unsafe.Pointer(cs))
+		cb := C.CString(b.thinkingSigBaseURL)
+		defer C.free(unsafe.Pointer(cb))
+		cm := C.CString(b.thinkingSigModel)
+		defer C.free(unsafe.Pointer(cm))
+		if C.panto_message_builder_add_thinking(p, ct, cs, C.PantoAPIStyle(b.thinkingSigStyle), cb, cm) != C.PANTO_OK {
+			return lastError()
+		}
+	case BlockToolUse:
+		cid := C.CString(b.toolID)
+		defer C.free(unsafe.Pointer(cid))
+		cn := C.CString(b.toolName)
+		defer C.free(unsafe.Pointer(cn))
+		ci := C.CString(b.input)
+		defer C.free(unsafe.Pointer(ci))
+		if C.panto_message_builder_add_tool_use(p, cid, cn, ci) != C.PANTO_OK {
+			return lastError()
+		}
+	case BlockToolResult:
+		cid := C.CString(b.toolID)
+		defer C.free(unsafe.Pointer(cid))
+		ct := C.CString(b.text)
+		defer C.free(unsafe.Pointer(ct))
+		if C.panto_message_builder_add_tool_result(p, cid, ct, C.bool(b.isError)) != C.PANTO_OK {
+			return lastError()
+		}
+	case BlockSystem:
+		ct := C.CString(b.text)
+		defer C.free(unsafe.Pointer(ct))
+		if C.panto_message_builder_add_system(p, ct, C.PantoSystemMode(b.systemMode)) != C.PANTO_OK {
+			return lastError()
+		}
+	default:
+		return fmt.Errorf("unknown content block tag %d", int(b.tag))
 	}
+	return nil
 }
 
-// AddMessage commits the builder's blocks as one message of the builder's role,
-// with optional assistant usage. It consumes the builder (success or failure);
-// the builder must not be used afterwards.
-func (c *Conversation) AddMessage(b *MessageBuilder, u *Usage) error {
+// AddMessage commits msg as one message of msg.Role (with its optional usage
+// and identity) — the block-level path for rebuilding a persisted conversation
+// containing tool calls or thinking that the text-only AddUserText/
+// AddAssistantText cannot.
+func (c *Conversation) AddMessage(msg InputMessage) error {
+	b, err := buildCBuilder(msg)
+	if err != nil {
+		return err
+	}
 	var cu C.PantoUsage
 	has := false
-	if u != nil {
-		cu = cUsage(*u)
+	if msg.Usage != nil {
+		cu = cUsage(*msg.Usage)
 		has = true
 	}
-	st := C.panto_conversation_add_message(c.ptr, b.ptr, C.bool(has), cu)
-	b.ptr = nil
-	runtime.SetFinalizer(b, nil)
-	if st != C.PANTO_OK {
+	// panto_conversation_add_message consumes (destroys) the builder on every
+	// outcome, so there is nothing to free here afterwards.
+	if C.panto_conversation_add_message(c.ptr, b, C.bool(has), cu) != C.PANTO_OK {
 		return lastError()
 	}
 	return nil
@@ -828,11 +936,21 @@ func (a *Agent) Compact(overrideSystemPrompt, extraInstructions *string) (Compac
 	}, nil
 }
 
-func (a *Agent) Run(userText string) (*Stream, error) {
-	ct := C.CString(userText)
-	defer C.free(unsafe.Pointer(ct))
+// Run opens a turn from a user-role message and returns a resumable Stream.
+// Build the message with UserMessage: a single TextBlock for a plain prompt,
+// or ToolResultBlocks to resume a turn whose tool calls the embedder handled
+// itself after breaking out of a prior stream. The message must be user-role.
+func (a *Agent) Run(msg InputMessage) (*Stream, error) {
+	if msg.Role != RoleUser {
+		return nil, fmt.Errorf("Run requires a user-role message, got role %d", int(msg.Role))
+	}
+	b, err := buildCBuilder(msg)
+	if err != nil {
+		return nil, err
+	}
+	// panto_agent_run consumes (destroys) the builder on every outcome.
 	var out *C.PantoStream
-	if C.panto_agent_run(a.ptr, ct, &out) != C.PANTO_OK {
+	if C.panto_agent_run(a.ptr, b, &out) != C.PANTO_OK {
 		return nil, lastError()
 	}
 	s := &Stream{ptr: out}
@@ -886,10 +1004,10 @@ func (s *Stream) Reopen() error {
 
 func (s *Stream) Next() (Event, bool, error) {
 	if s.err != nil {
-		return Event{}, false, s.err
+		return nil, false, s.err
 	}
 	if s.done {
-		return Event{}, false, nil
+		return nil, false, nil
 	}
 	var cev C.PantoEvent
 	status := C.panto_stream_next(s.ptr, &cev)
@@ -897,19 +1015,19 @@ func (s *Stream) Next() (Event, bool, error) {
 	case C.PANTO_NEXT_EVENT:
 		defer C.panto_event_free(&cev)
 		ev := marshalEvent(&cev)
-		if ev.Tag == EventTurnComplete {
+		if ev.EventType() == EventTurnComplete {
 			s.done = true
 		}
 		return ev, true, nil
 	case C.PANTO_NEXT_DONE:
 		s.done = true
-		return Event{}, false, nil
+		return nil, false, nil
 	case C.PANTO_NEXT_ERROR:
 		s.err = lastError()
-		return Event{}, false, s.err
+		return nil, false, s.err
 	default:
 		s.err = fmt.Errorf("unknown panto next status %d", int(status))
-		return Event{}, false, s.err
+		return nil, false, s.err
 	}
 }
 func (s *Stream) Err() error { return s.err }
@@ -923,7 +1041,7 @@ func (s *Stream) Iter() iter.Seq[Event] {
 			if !yield(ev) {
 				return
 			}
-			if ev.Tag == EventTurnComplete {
+			if ev.EventType() == EventTurnComplete {
 				return
 			}
 		}
@@ -1021,31 +1139,36 @@ func setAnthropic(out *C.PantoProviderConfig, cfg *AnthropicMessagesConfig, cs *
 }
 
 func marshalEvent(cev *C.PantoEvent) Event {
-	ev := Event{Tag: EventTag(cev.tag)}
 	switch cev.tag {
 	case C.PANTO_EVENT_MESSAGE_START:
-		ev.MessageStart = MessageRole(*(*C.PantoMessageRole)(unsafe.Pointer(&cev.data[0])))
+		return MessageStartEvent{Role: MessageRole(*(*C.PantoMessageRole)(unsafe.Pointer(&cev.data[0])))}
 	case C.PANTO_EVENT_BLOCK_START:
 		b := (*C.PantoBlockStart)(unsafe.Pointer(&cev.data[0]))
-		ev.BlockStart = BlockStart{BlockType: ContentBlockTag(b.block_type), Index: uint(b.index)}
+		return BlockStartEvent{BlockType: ContentBlockTag(b.block_type), Index: uint(b.index)}
 	case C.PANTO_EVENT_TOOL_DETAILS:
 		t := (*C.PantoToolDetails)(unsafe.Pointer(&cev.data[0]))
-		ev.ToolDetails = ToolDetails{Index: uint(t.index), ID: goSlice(t.id), Name: goSlice(t.name)}
+		return ToolDetailsEvent{Index: uint(t.index), ID: goSlice(t.id), Name: goSlice(t.name)}
 	case C.PANTO_EVENT_CONTENT_DELTA:
 		d := (*C.PantoContentDelta)(unsafe.Pointer(&cev.data[0]))
-		ev.ContentDelta = ContentDelta{Index: uint(d.index), Delta: goSlice(d.delta)}
+		return ContentDeltaEvent{Index: uint(d.index), Delta: goSlice(d.delta)}
 	case C.PANTO_EVENT_BLOCK_COMPLETE:
 		b := (*C.PantoBlockComplete)(unsafe.Pointer(&cev.data[0]))
-		ev.BlockComplete = BlockComplete{Index: uint(b.index), BlockType: ContentBlockTag(b.block_type)}
+		return BlockCompleteEvent{Index: uint(b.index), BlockType: ContentBlockTag(b.block_type)}
 	case C.PANTO_EVENT_MESSAGE_COMPLETE:
 		m := (*C.PantoMessageComplete)(unsafe.Pointer(&cev.data[0]))
+		ev := MessageCompleteEvent{
+			Role:         MessageRole(m.role),
+			MessageIndex: uint(m.message_index),
+			BlockCount:   uint(m.block_count),
+		}
 		if m.has_usage {
 			u := usage(m.usage)
-			ev.MessageComplete.Usage = &u
+			ev.Usage = &u
 		}
+		return ev
 	case C.PANTO_EVENT_PROVIDER_RETRY:
 		r := (*C.PantoProviderRetry)(unsafe.Pointer(&cev.data[0]))
-		ev.ProviderRetry = ProviderRetry{
+		ev := ProviderRetryEvent{
 			Attempt:     uint(r.attempt),
 			MaxAttempts: uint(r.max_attempts),
 			DelayMS:     uint64(r.delay_ms),
@@ -1055,20 +1178,23 @@ func marshalEvent(cev *C.PantoEvent) Event {
 		}
 		if r.has_status_code {
 			v := uint16(r.status_code)
-			ev.ProviderRetry.StatusCode = &v
+			ev.StatusCode = &v
 		}
 		if r.has_retry_after_ms {
 			v := uint64(r.retry_after_ms)
-			ev.ProviderRetry.RetryAfterMS = &v
+			ev.RetryAfterMS = &v
 		}
+		return ev
 	case C.PANTO_EVENT_TOOL_DISPATCH_START:
 		t := (*C.PantoToolDispatchStart)(unsafe.Pointer(&cev.data[0]))
-		ev.ToolDispatchStart = ToolDispatchStart{Count: uint(t.count)}
+		return ToolDispatchStartEvent{Count: uint(t.count)}
 	case C.PANTO_EVENT_TOOL_DISPATCH_COMPLETE:
 		t := (*C.PantoToolDispatchComplete)(unsafe.Pointer(&cev.data[0]))
-		ev.ToolDispatchComplete = ToolDispatchComplete{MessageIndex: uint(t.message_index)}
+		return ToolDispatchCompleteEvent{MessageIndex: uint(t.message_index)}
+	case C.PANTO_EVENT_TURN_COMPLETE:
+		return TurnCompleteEvent{}
 	}
-	return ev
+	return TurnCompleteEvent{}
 }
 
 func goSlice(s C.PantoSlice) string {
diff --git a/libpanto-go/panto_test.go b/libpanto-go/panto_test.go
index fd9df70..af8cafc 100644
--- a/libpanto-go/panto_test.go
+++ b/libpanto-go/panto_test.go
@@ -59,7 +59,7 @@ func TestCustomStoreMintsSession(t *testing.T) {
 }
 
 // Gap #3: rebuild a conversation containing tool_use / tool_result / thinking
-// via MessageBuilder, then read the blocks back losslessly.
+// via the value-block API, then read the blocks back losslessly.
 func TestMessageBuilderRoundTrip(t *testing.T) {
 	Init()
 	defer Deinit()
@@ -70,31 +70,17 @@ func TestMessageBuilderRoundTrip(t *testing.T) {
 	}
 	defer conv.Close()
 
-	ub, err := NewMessageBuilder(RoleUser)
-	if err != nil {
-		t.Fatalf("NewMessageBuilder(user): %v", err)
-	}
-	if err := ub.AddToolResult("call-1", "result text", false); err != nil {
-		t.Fatalf("AddToolResult: %v", err)
-	}
-	if err := conv.AddMessage(ub, nil); err != nil {
+	if err := conv.AddMessage(UserMessage(
+		ToolResultBlock("call-1", "result text", false),
+	)); err != nil {
 		t.Fatalf("AddMessage(user): %v", err)
 	}
 
-	ab, err := NewMessageBuilder(RoleAssistant)
-	if err != nil {
-		t.Fatalf("NewMessageBuilder(assistant): %v", err)
-	}
-	if err := ab.AddText("let me search"); err != nil {
-		t.Fatalf("AddText: %v", err)
-	}
-	if err := ab.AddThinking("hmm", "sig", AnthropicMessages, "https://api", "claude"); err != nil {
-		t.Fatalf("AddThinking: %v", err)
-	}
-	if err := ab.AddToolUse("call-1", "search", `{"q":1}`); err != nil {
-		t.Fatalf("AddToolUse: %v", err)
-	}
-	if err := conv.AddMessage(ab, &Usage{Input: 10, Output: 5}); err != nil {
+	if err := conv.AddMessage(AssistantMessage(&Usage{Input: 10, Output: 5},
+		TextBlock("let me search"),
+		ThinkingBlock("hmm", "sig", AnthropicMessages, "https://api", "claude"),
+		ToolUseBlock("call-1", "search", `{"q":1}`),
+	)); err != nil {
 		t.Fatalf("AddMessage(assistant): %v", err)
 	}
 
@@ -141,22 +127,18 @@ func TestMessageBuilderSetIdentity(t *testing.T) {
 	}
 	defer conv.Close()
 
-	ab, err := NewMessageBuilder(RoleAssistant)
-	if err != nil {
-		t.Fatalf("NewMessageBuilder(assistant): %v", err)
-	}
-	// Thinking block with a signature but NO per-block origin.
-	if err := ab.AddThinking("reasoned", "sig", OpenAIChat, "", ""); err != nil {
-		t.Fatalf("AddThinking: %v", err)
-	}
-	if err := ab.AddText("answer"); err != nil {
-		t.Fatalf("AddText: %v", err)
-	}
-	// Provenance is carried at the message level instead.
-	if err := ab.SetIdentity(AnthropicMessages, "https://api.anthropic.com", "claude", ReasoningDefault); err != nil {
-		t.Fatalf("SetIdentity: %v", err)
-	}
-	if err := conv.AddMessage(ab, &Usage{Input: 10, Output: 5}); err != nil {
+	// Thinking block with a signature but NO per-block origin; provenance is
+	// carried at the message level instead, via WithIdentity.
+	msg := AssistantMessage(&Usage{Input: 10, Output: 5},
+		ThinkingBlock("reasoned", "sig", OpenAIChat, "", ""),
+		TextBlock("answer"),
+	).WithIdentity(MessageIdentity{
+		APIStyle:  AnthropicMessages,
+		BaseURL:   "https://api.anthropic.com",
+		Model:     "claude",
+		Reasoning: ReasoningDefault,
+	})
+	if err := conv.AddMessage(msg); err != nil {
 		t.Fatalf("AddMessage: %v", err)
 	}
 
diff --git a/libpanto-lua/src/module.zig b/libpanto-lua/src/module.zig
index 349a59e..ecf89af 100644
--- a/libpanto-lua/src/module.zig
+++ b/libpanto-lua/src/module.zig
@@ -602,7 +602,13 @@ fn agentRun(L_opt: ?*c.lua_State) callconv(.c) c_int {
     var len: usize = 0;
     const ptr = c.luaL_checklstring(L, 2, &len);
 
-    const stream = box.agent.run(.{ .text = ptr[0..len] }) catch
+    // Open the turn from a single user text block; `run` adopts the block.
+    const alloc = box.agent.conversation.allocator;
+    var blocks = [_]panto.ContentBlock{
+        .{ .Text = panto.textualBlockFromSlice(alloc, ptr[0..len]) catch
+            return luaErr(L, "panto: out of memory") },
+    };
+    const stream = box.agent.run(.{ .blocks = &blocks }) catch
         return luaErr(L, "panto: failed to start turn");
 
     const ud = c.lua_newuserdatauv(L, @sizeOf(StreamBox), 0) orelse {
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig
index ed1600f..e689924 100644
--- a/libpanto/src/agent.zig
+++ b/libpanto/src/agent.zig
@@ -200,11 +200,14 @@ fn toolErrorResult(
     return tool_mod.ResultParts.fromTextOwned(allocator, msg);
 }
 
-/// The user's submission that opens a turn. A struct (not a bare slice) so
-/// it can grow to carry file/image attachments alongside the chat text
-/// without changing `Agent.run`'s signature.
+/// The user's submission that opens a turn: an ordered list of content
+/// blocks (text, tool results, images, …), the same shape used to rebuild a
+/// persisted user message. Ownership of `blocks` transfers to the agent's
+/// conversation on `run`; the caller must not deinit them afterwards. A plain
+/// chat turn is one `.Text` block; a turn that resumes after the embedder
+/// handled tool calls is one or more `.ToolResult` blocks.
 pub const UserMessage = struct {
-    text: []const u8,
+    blocks: []const conversation.ContentBlock,
 };
 
 /// Outcome of a compaction attempt.
@@ -259,6 +262,14 @@ pub const Agent = struct {
     /// touched from the single agent-loop thread (retries are serial), so
     /// no synchronization is needed.
     _retry_prng: ?std.Random.DefaultPrng = null,
+    /// High-water mark of messages durably handed to the store: the count of
+    /// conversation messages already persisted. `flushPersist` appends only
+    /// `[_persisted_through .. coherent_end)` and advances this, so the store
+    /// stays current at every `Stream.next()` boundary without re-appending.
+    /// Initialized to the loaded history length (already persisted) and
+    /// rewound by `rewriteWithSummary` so a compaction re-persists its summary
+    /// + restated suffix.
+    _persisted_through: usize = 0,
 
     /// Construct an agent.
     ///
@@ -286,6 +297,9 @@ pub const Agent = struct {
             .conversation = maybe_conversation orelse conversation.Conversation.init(allocator),
             ._session = session,
         };
+        // Loaded history is already in the store; only messages produced from
+        // here on need persisting.
+        self._persisted_through = self.conversation.messages.items.len;
         return self;
     }
 
@@ -353,19 +367,11 @@ pub const Agent = struct {
         text: []const u8,
         mode: conversation.SystemMode,
     ) !void {
-        const start = self.conversation.messages.items.len;
         switch (mode) {
             .append => try self.conversation.addSystemMessage(text),
             .replace => try self.conversation.replaceSystemMessage(text),
         }
-        try turn_persist.persistTurn(
-            self._allocator,
-            &self._session,
-            &self.conversation,
-            start,
-            self.wireIdentity(),
-            &.{},
-        );
+        self.flushPersist();
     }
 
     /// Submit a user message and begin a turn, returning a resumable pull
@@ -390,49 +396,55 @@ pub const Agent = struct {
         self._auto_compacted = false;
 
         // Append + persist the user prompt up front (the dangling-prompt
-        // recovery guarantee).
-        const user_start = self.conversation.messages.items.len;
-        try addUserText(&self.conversation, message.text);
-        try turn_persist.persistTurn(
-            self._allocator,
-            &self._session,
-            &self.conversation,
-            user_start,
-            self.wireIdentity(),
-            &.{},
-        );
+        // recovery guarantee). `addMessage` adopts the blocks; persistence is
+        // brought current through the shared high-water flush.
+        try self.conversation.addMessage(.user, message.blocks, null);
+        self.flushPersist();
 
         const s = try self._allocator.create(Stream);
         s.* = Stream.init(self);
         return s;
     }
 
-    /// Persist the messages a turn produced. When the turn auto-compacted,
-    /// message indices shifted (the conversation was rewritten to
-    /// `[system..., summary, kept-suffix...]`), so persist the whole
-    /// post-compaction window instead of `[start..]`.
-    fn persistTurnTail(self: *Agent, start: usize) !void {
-        const id = self.wireIdentity();
-        if (self._auto_compacted) {
-            try turn_persist.persistCompaction(
-                self._allocator,
-                &self._session,
-                &self.conversation,
-                id,
-                &.{},
-            );
-        } else {
-            try turn_persist.persistTurn(
-                self._allocator,
-                &self._session,
-                &self.conversation,
-                start,
-                id,
-                &.{},
-            );
+    /// Bring the session store current with the conversation: append every
+    /// message in `[_persisted_through .. coherent_end)` as a single batch and
+    /// advance the high-water mark. `coherent_end` excludes a trailing
+    /// assistant message whose ToolUse blocks have no following results — a
+    /// turn interrupted between the assistant's tool call and dispatch. Leaving
+    /// it unpersisted keeps the log replayable and the live conversation in a
+    /// state `compact()` (or a tool-result `run`) can resume from. Called at
+    /// every `Stream.next()` boundary, so after any `next()` the store reflects
+    /// all committed, coherent messages.
+    pub fn flushPersist(self: *Agent) void {
+        const msgs = self.conversation.messages.items;
+        // An operation that rewrote the prefix (replace-system, cancel) can
+        // leave the mark past the new length; clamp before comparing.
+        if (self._persisted_through > msgs.len) self._persisted_through = msgs.len;
+        var end = msgs.len;
+        if (end > self._persisted_through and
+            msgs[end - 1].role == .assistant and
+            turn_persist.hasToolUseWithoutFollowingResults(&self.conversation, end - 1))
+        {
+            end -= 1;
         }
+        if (end <= self._persisted_through) return;
+        turn_persist.persistRange(
+            self._allocator,
+            &self._session,
+            &self.conversation,
+            self._persisted_through,
+            end,
+            self.wireIdentity(),
+            &.{},
+        ) catch |e| {
+            std.log.err("session: failed to persist turn: {t}", .{e});
+            return;
+        };
+        self._persisted_through = end;
     }
 
+    /// Persist the messages a turn produced. When the turn auto-compacted,
+    /// message indices shifted (the conversation was rewritten to
     fn hasToolUseBlock(msg: conversation.Message) bool {
         for (msg.content.items) |block| {
             if (block == .ToolUse) return true;
@@ -653,15 +665,9 @@ pub const Agent = struct {
             self._config.compaction.compaction_prompt orelse
             return error.NoCompactionPrompt;
         const res = try self._compactInPlace(system_prompt, extra_instructions);
-        if (res.compacted) {
-            try turn_persist.persistCompaction(
-                self._allocator,
-                &self._session,
-                &self.conversation,
-                self.wireIdentity(),
-                &.{},
-            );
-        }
+        // `rewriteWithSummary` rewound the high-water mark to the summary;
+        // flush appends the new compaction window (summary + restated suffix).
+        if (res.compacted) self.flushPersist();
         return res;
     }
 
@@ -752,6 +758,13 @@ pub const Agent = struct {
         for (conv.messages.items) |*m| m.deinit(alloc);
         conv.messages.deinit(alloc);
         conv.messages = rebuilt;
+
+        // The rewrite invalidated the old message indices. Rewind the
+        // persistence high-water mark to the inserted summary so the next
+        // `flushPersist` re-appends the new compaction window (summary +
+        // restated kept suffix) as fresh entries.
+        self._persisted_through =
+            conversation.latestCompactionIndex(conv.messages.items) orelse 0;
     }
 
     /// Run a single compaction provider call against a throwaway
@@ -1122,10 +1135,8 @@ pub const Stream = struct {
     state: State,
     /// The active provider response, when in `.streaming`.
     _response: ?provider_mod.ProviderStream = null,
-    /// First message index of this turn (for persistence).
+    /// First message index of this turn (the boundary `cancel` rolls back to).
     _start: usize,
-    /// Set once the turn's tail has been persisted (on terminal or deinit).
-    _persisted: bool = false,
     /// A terminal error to surface once any already-queued events (e.g.
     /// `provider_retry` notices pushed before the failing attempt) have been
     /// drained. `next()` yields the queue first, then this error.
@@ -1173,7 +1184,9 @@ pub const Stream = struct {
     pub fn deinit(self: *Stream) void {
         // Persist whatever the turn committed, on every exit path — including
         // dropping the stream mid-turn after some messages were committed.
-        self.persistTail();
+        // `next()` already flushes at every boundary, so this is usually a
+        // no-op; it catches a stream dropped without a final `next()`.
+        self._agent.flushPersist();
         if (self._response) |ps| ps.deinit();
         if (self._retry_message) |m| self._agent._allocator.free(m);
         self._queue.deinit();
@@ -1197,17 +1210,11 @@ pub const Stream = struct {
             var msg = conv.messages.pop().?;
             msg.deinit(conv.allocator);
         }
+        // Rolled the conversation back; clamp the persistence mark to match.
+        self._agent.flushPersist();
         self.state = .done;
     }
 
-    fn persistTail(self: *Stream) void {
-        if (self._persisted) return;
-        self._persisted = true;
-        self._agent.persistTurnTail(self._start) catch |e| {
-            std.log.err("session: failed to persist turn: {t}", .{e});
-        };
-    }
-
     /// Reset a `.failed` stream back to `.turn_start` so the caller can resume
     /// `next()` after changing the agent config (e.g. an embedder that catches
     /// a terminal `ProviderBadRequest`, rewrites the provider config, and
@@ -1230,6 +1237,11 @@ pub const Stream = struct {
     /// Pull the next event, or null past the terminal. See the contract
     /// above.
     pub fn next(self: *Stream) !?Event {
+        // Bring persistence current on every return path (events, terminal,
+        // errors). This is the interruptibility guarantee: after any `next()`
+        // the store reflects all committed, coherent messages, so the caller
+        // can `break` and immediately `compact()` or resume with a new turn.
+        defer self._agent.flushPersist();
         // Always drain queued events first; they borrow decode/conversation
         // state valid until this call returns.
         if (self._queue.pop()) |ev| return ev;
@@ -1349,7 +1361,6 @@ pub const Stream = struct {
 
                     if (!Agent.hasToolUseBlock(last)) {
                         self.state = .done;
-                        self.persistTail();
                         return .turn_complete;
                     }
 
@@ -1571,11 +1582,21 @@ const testing = std.testing;
 /// `submitUserMessage` + `runStep`: it returns the same terminal error a
 /// turn would raise.
 fn drainTurn(agent: *Agent, text: []const u8) !void {
-    var s = try agent.run(.{ .text = text });
+    var s = try runUserText(agent, text);
     defer s.deinit();
     while (try s.next()) |_| {}
 }
 
+/// Test helper: open a turn from a single user `.Text` block, mirroring the
+/// old text-only `run(.{ .text = ... })`. The block is adopted by the agent's
+/// conversation.
+fn runUserText(agent: *Agent, text: []const u8) !*Stream {
+    var blocks = [_]conversation.ContentBlock{
+        .{ .Text = try conversation.textualBlockFromSlice(agent._allocator, text) },
+    };
+    return agent.run(.{ .blocks = &blocks });
+}
+
 /// Test helper: the items of a ToolResultBlock's first text part.
 fn trText(tr: conversation.ToolResultBlock) []const u8 {
     for (tr.parts.items) |p| {
@@ -2102,6 +2123,78 @@ test "agent persists user, assistant, and tool-result messages of a turn" {
     }
 }
 
+test "interruption: persistence stays current and excludes a dangling tool call" {
+    const allocator = testing.allocator;
+
+    const scripted = [_]StubProvider.ScriptedTurn{
+        .{ .blocks = &.{
+            .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } },
+        } },
+        .{ .blocks = &.{
+            .{ .Text = "ok" },
+        } },
+    };
+    var stub = StubProvider{ .allocator = allocator, .scripted = &scripted };
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var h = TestHarness.init(allocator);
+    defer h.deinit();
+    try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:"));
+    h.activate();
+
+    var cap = CapturingStore.init(allocator);
+    defer cap.deinit();
+    const agent = try Agent.init(allocator, io, &h.config, cap.store().create(), null);
+    defer agent.deinit();
+    h.seedInto(agent);
+    agent._open_stream_fn = stub.install();
+
+    var s = try runUserText(agent, "call a tool");
+
+    // `run` persisted the user prompt up front, and nothing else yet.
+    try testing.expectEqual(@as(usize, 1), cap.roles.items.len);
+    try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[0]);
+
+    // Pull until the assistant's tool-call message completes, then break — the
+    // embedder's interruption point (e.g. to handle the tool calls itself).
+    while (try s.next()) |ev| {
+        if (ev == .message_complete) break;
+    }
+
+    // The tool-call message is live in the conversation...
+    const msgs = agent.conversation.messages.items;
+    try testing.expectEqual(conversation.MessageRole.assistant, msgs[msgs.len - 1].role);
+    try testing.expect(Agent.hasToolUseBlock(msgs[msgs.len - 1]));
+    // ...but it is a dangling tool call (no results), so the store still holds
+    // only the coherent prefix: the interruption guarantee.
+    try testing.expectEqual(@as(usize, 1), cap.roles.items.len);
+
+    // Dropping the stream mid-turn must not persist the dangling call either.
+    s.deinit();
+    try testing.expectEqual(@as(usize, 1), cap.roles.items.len);
+
+    // The conversation is coherent enough to resume from: feeding the tool
+    // result as a fresh user turn resolves the dangling call, and now both the
+    // assistant tool-call and the user result become persistable.
+    var parts: std.ArrayList(conversation.ResultPartStored) = .empty;
+    try parts.append(allocator, .{ .text = try conversation.textualBlockFromSlice(allocator, "42") });
+    var s2 = try agent.run(.{ .blocks = &.{
+        .{ .ToolResult = .{
+            .tool_use_id = try allocator.dupe(u8, "tc_1"),
+            .parts = parts,
+            .is_error = false,
+        } },
+    } });
+    defer s2.deinit();
+    // user prompt, assistant(ToolUse), user(ToolResult) are all persisted once
+    // the dangling call is resolved by the new user turn.
+    try testing.expectEqual(@as(usize, 3), cap.roles.items.len);
+    try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[0]);
+    try testing.expectEqual(conversation.MessageRole.assistant, cap.roles.items[1]);
+    try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[2]);
+}
+
 test "agent runs a turn against NullStore without persisting or erroring" {
     const allocator = testing.allocator;
 
@@ -2455,7 +2548,7 @@ test "Stream emits tool_dispatch_start before running tools" {
     h.seedInto(agent);
     agent._open_stream_fn = stub.install();
 
-    var s = try agent.run(.{ .text = "call a tool" });
+    var s = try runUserText(agent, "call a tool");
     defer s.deinit();
 
     const first = (try s.next()).?;
@@ -2917,6 +3010,106 @@ test "compact: summarizes prefix, keeps suffix, system survives" {
     );
 }
 
+test "interruption: resuming next() after a break dispatches the pending tools" {
+    const allocator = testing.allocator;
+
+    const scripted = [_]StubProvider.ScriptedTurn{
+        .{ .blocks = &.{
+            .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } },
+        } },
+        .{ .blocks = &.{
+            .{ .Text = "ok" },
+        } },
+    };
+    var stub = StubProvider{ .allocator = allocator, .scripted = &scripted };
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var h = TestHarness.init(allocator);
+    defer h.deinit();
+    try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:"));
+    h.activate();
+
+    var cap = CapturingStore.init(allocator);
+    defer cap.deinit();
+    const agent = try Agent.init(allocator, io, &h.config, cap.store().create(), null);
+    defer agent.deinit();
+    h.seedInto(agent);
+    agent._open_stream_fn = stub.install();
+
+    var s = try runUserText(agent, "call a tool");
+    defer s.deinit();
+
+    // Break right after the assistant's tool-call message completes — the
+    // stream is paused, NOT closed. The tool has not been dispatched.
+    while (try s.next()) |ev| {
+        if (ev == .message_complete) break;
+    }
+    try testing.expectEqual(@as(usize, 1), cap.roles.items.len); // only the prompt
+
+    // Resuming `next()` on the same stream picks up where it left off and runs
+    // the agent's registered-tool machinery: no new user turn required.
+    var saw_dispatch = false;
+    while (try s.next()) |ev| {
+        if (ev == .tool_dispatch_complete) saw_dispatch = true;
+    }
+    try testing.expect(saw_dispatch);
+
+    // The full turn is now persisted: prompt, assistant(ToolUse),
+    // user(ToolResult), assistant(text).
+    try testing.expectEqual(@as(usize, 4), cap.roles.items.len);
+    try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[2]);
+}
+
+test "compact: tolerates a trailing dangling tool call (interrupted turn)" {
+    const allocator = testing.allocator;
+
+    const scripted = [_]StubProvider.ScriptedTurn{
+        .{ .blocks = &.{.{ .Text = "SUMMARY OF EARLIER" }} },
+    };
+    var stub = StubProvider{ .allocator = allocator, .scripted = &scripted };
+    var threaded: std.Io.Threaded = .init(allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var h = TestHarness.init(allocator);
+    defer h.deinit();
+    h.activate();
+    h.config.compaction = .{ .keep_verbatim = 10 };
+    var ns = null_store_mod.NullStore.init(allocator);
+    const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null);
+    defer agent.deinit();
+    h.seedInto(agent);
+    agent._open_stream_fn = stub.install();
+
+    const conv = &agent.conversation;
+    try conv.addSystemMessage("you are helpful");
+    try addUserText(conv, "first question here with several words");
+    try conv.addAssistantMessage(&.{
+        .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") },
+    }, null);
+    // A fresh turn whose assistant message ends in a tool call with no results
+    // yet — exactly the state a break right after `message_complete` leaves.
+    try addUserText(conv, "second recent question");
+    try conv.addAssistantMessage(&.{
+        .{ .ToolUse = .{
+            .id = try allocator.dupe(u8, "tc_1"),
+            .name = try allocator.dupe(u8, "echo"),
+            .input = try conversation.textualBlockFromSlice(allocator, "hello"),
+        } },
+    }, null);
+
+    // Compaction must not choke on the dangling tail: it summarizes the older
+    // turn and keeps the interrupted turn (incl. the dangling call) verbatim.
+    const res = try agent._compactInPlace("Summarize the conversation.", null);
+    try testing.expect(res.compacted);
+
+    // The dangling tool call survives as the tail, intact and resumable.
+    const last = conv.messages.items[conv.messages.items.len - 1];
+    try testing.expectEqual(conversation.MessageRole.assistant, last.role);
+    try testing.expect(Agent.hasToolUseBlock(last));
+    try testing.expectEqualStrings("tc_1", last.content.items[0].ToolUse.id);
+}
+
 test "compact: restated suffix usage reconstructs a fresh cumulative chain" {
     const allocator = testing.allocator;
 
@@ -3175,7 +3368,7 @@ const RetryRecorder = struct {
 /// `provider_retry` event into `rr` and discarding the rest. Returns the
 /// same terminal error the turn would raise.
 fn drainTurnRecording(agent: *Agent, text: []const u8, rr: *RetryRecorder) !void {
-    var s = try agent.run(.{ .text = text });
+    var s = try runUserText(agent, text);
     defer s.deinit();
     while (try s.next()) |ev| {
         if (ev == .provider_retry) try rr.append(ev.provider_retry);
diff --git a/libpanto/src/turn_persist.zig b/libpanto/src/turn_persist.zig
index 1dacfec..6d260c8 100644
--- a/libpanto/src/turn_persist.zig
+++ b/libpanto/src/turn_persist.zig
@@ -46,12 +46,28 @@ pub fn persistTurn(
     identity: WireIdentity,
     tools: []const ToolDecl,
 ) !void {
+    return persistRange(alloc, session, conv, start_index, conv.messages.items.len, identity, tools);
+}
+
+/// Persist conversation messages in `[start_index, end_index)`. Like
+/// `persistTurn` but with an explicit upper bound, so an incremental flush can
+/// exclude a trailing not-yet-coherent message (a dangling tool call awaiting
+/// its results) while still committing everything before it.
+pub fn persistRange(
+    alloc: Allocator,
+    session: *Session,
+    conv: *conversation.Conversation,
+    start_index: usize,
+    end_index: usize,
+    identity: WireIdentity,
+    tools: []const ToolDecl,
+) !void {
     var batch: std.ArrayList(PersistentMessage) = .empty;
     defer batch.deinit(alloc);
 
     const all_messages = conv.messages.items;
     var i = start_index;
-    while (i < all_messages.len) : (i += 1) {
+    while (i < end_index) : (i += 1) {
         const msg = &all_messages[i];
         if (msg.role == .assistant and hasToolUseWithoutFollowingResults(conv, i)) {
             continue;
diff --git a/src/tui_app.zig b/src/tui_app.zig
index 4313556..88c2c66 100644
--- a/src/tui_app.zig
+++ b/src/tui_app.zig
@@ -1947,7 +1947,7 @@ fn handleSubmittedLine(app: *App, term: *Terminal, line: []const u8, opts: RunOp
         return;
     };
 
-    driveTurn(app, term, opts, .{ .text = line }) catch |err| {
+    driveTurn(app, term, opts, line) catch |err| {
         try app.routeError(err);
     };
     try app.renderNow();
@@ -2025,8 +2025,12 @@ fn tryAdaptiveFallback(app: *App, err: anyerror) bool {
 /// component state until it terminates, rendering coalesced frames as deltas
 /// arrive. The stream is always `deinit`ed (persisting the turn tail) on every
 /// exit path — agent persistence is untouched.
-fn driveTurn(app: *App, term: *Terminal, opts: RunOptions, message: panto.UserMessage) !void {
-    var stream = try opts.agent.run(message);
+fn driveTurn(app: *App, term: *Terminal, opts: RunOptions, message_text: []const u8) !void {
+    // Open the turn from a single user text block; `run` adopts the block.
+    var blocks = [_]panto.ContentBlock{
+        .{ .Text = try panto.textualBlockFromSlice(app.alloc, message_text) },
+    };
+    var stream = try opts.agent.run(.{ .blocks = &blocks });
     defer stream.deinit();
     // Two single-shot fallbacks can re-open the SAME turn (no user-message
     // duplication):