Expose recovery view for assistant-prefix decisions

threads/EXECUTOR_RECOVERY.md

@@ -230,6 +230,8 @@ Current status:
 - landed for streamer reporting
   - `LLMStreamer` exposes `Capabilities()`
   - current streamers report `AssistantPrefix`
+- landed for recovery-view surfacing
+  - `Thread.RecoveryView()` exposes exact-recovery capability requirements
 - attach-time recovery policy still does not use the capability surface yet
 
 ## Tool Recovery Model
@@ -388,12 +390,17 @@ What this already gives us:
 
 What is still missing from this derived view for recovery:
 
-- durable per-call recovery metadata for started calls
-- continuation mode in the derived view
-- an explicit recovery-facing API if attach-time recovery should live outside
-  `thread.go`
+- policy application on top of the derived view
 
-The exact API shape can still change. The important capability now exists.
+Current status:
+
+- `Thread.RecoveryView()` exposes:
+  - outstanding tool calls
+  - handler load data binding
+  - started state
+  - continuation mode
+  - durable recovery metadata
+  - exact-recovery streamer capability requirements
 
 ### Binding Resolution Rule
 

threads/README.md

@@ -32,6 +32,8 @@ implemented feature overview.
   - queueing `SendItem{}` moves the thread into request construction
   - `LLMStreamer.Capabilities()` reports provider request-shape support such as
     assistant-prefixed continuation
+  - `Thread.RecoveryView()` exposes the exact-recovery capability requirements
+    and outstanding tool-call shape for attach-time recovery decisions
   - streamed items are appended back onto the thread in order
   - `ToolCallChunk` values are accumulated by call id until a final `ToolCall`
     arrives

threads/controlblock.go

@@ -28,11 +28,12 @@ type cbTransition struct {
 }
 
 type pendingToolCall struct {
-	call     ToolCall
-	load     json.RawMessage
-	started  bool
-	bound    bool
-	recovery ToolRecovery
+	call         ToolCall
+	load         json.RawMessage
+	started      bool
+	bound        bool
+	continueMode ToolContinue
+	recovery     ToolRecovery
 }
 
 type cbStateHandler interface {
@@ -438,6 +439,7 @@ func (cb *controlBlock) pendingToolCalls(items cbItems) []pendingToolCall {
 			call := pendingToolCall{call: v, load: append(json.RawMessage(nil), load...), bound: ok}
 			if i, ok := pendingByID[v.CallID]; ok {
 				call.started = pending[i].started
+				call.continueMode = pending[i].continueMode
 				call.recovery = pending[i].recovery
 				pending[i] = call
 				break
@@ -447,6 +449,7 @@ func (cb *controlBlock) pendingToolCalls(items cbItems) []pendingToolCall {
 		case ToolCallStarted:
 			if i, ok := pendingByID[v.CallID]; ok {
 				pending[i].started = true
+				pending[i].continueMode = v.Continue
 				pending[i].recovery = v.Recovery
 			}
 		case ToolCallResultable:

threads/recovery_view.go

@@ -0,0 +1,58 @@
+package threads
+
+import "encoding/json"
+
+// RecoveryView exposes the recovery-relevant shape of a thread without applying
+// any recovery policy.
+type RecoveryView struct {
+	State                 State
+	ExactRecoveryRequires StreamerCapabilities
+	OutstandingToolCalls  []OutstandingToolCall
+}
+
+// CanRecoverExactlyWith reports whether the provided streamer capabilities can
+// continue from the thread's exact retained state.
+func (v RecoveryView) CanRecoverExactlyWith(caps StreamerCapabilities) bool {
+	if v.ExactRecoveryRequires.AssistantPrefix && !caps.AssistantPrefix {
+		return false
+	}
+	return true
+}
+
+// OutstandingToolCall is the recovery-facing view of an unresolved tool call.
+type OutstandingToolCall struct {
+	Call            ToolCall
+	HandlerLoadData json.RawMessage
+	Started         bool
+	Bound           bool
+	Continue        ToolContinue
+	Recovery        ToolRecovery
+}
+
+// RecoveryView returns the recovery-relevant shape of the current thread.
+func (t *Thread) RecoveryView() RecoveryView {
+	pending := t.cb.pendingToolCalls(&t.items)
+	view := RecoveryView{
+		State:                 t.State(),
+		ExactRecoveryRequires: exactRecoveryRequirements(t.State()),
+		OutstandingToolCalls:  make([]OutstandingToolCall, 0, len(pending)),
+	}
+	for _, p := range pending {
+		view.OutstandingToolCalls = append(view.OutstandingToolCalls, OutstandingToolCall{
+			Call:            p.call,
+			HandlerLoadData: append(json.RawMessage(nil), p.load...),
+			Started:         p.started,
+			Bound:           p.bound,
+			Continue:        p.continueMode,
+			Recovery:        p.recovery,
+		})
+	}
+	return view
+}
+
+func exactRecoveryRequirements(state State) StreamerCapabilities {
+	if state == StateReceivingStream {
+		return StreamerCapabilities{AssistantPrefix: true}
+	}
+	return StreamerCapabilities{}
+}

threads/recovery_view_test.go

@@ -0,0 +1,126 @@
+package threads
+
+import (
+	"testing"
+)
+
+func TestRecoveryViewExposesOutstandingToolCallMetadata(t *testing.T) {
+	thread := newTestThread(t)
+	thread.QueueItem(ToolsSnapshot{Handlers: []ToolHandlerBinding{{
+		Name:            "calc",
+		HandlerLoadData: []byte(`{"snapshot":"bound"}`),
+	}}})
+	thread.QueueItem(ToolCall{CallID: "c1", Name: "missing", Payload: `{"a":1}`})
+	thread.QueueItem(ToolCall{CallID: "c2", Name: "calc", Payload: `{"a":2}`})
+	thread.QueueItem(ToolCallStarted{
+		CallID:   "c2",
+		Continue: ToolContinueManual,
+		Recovery: ToolRecoveryUnsafe,
+	})
+
+	got := thread.RecoveryView()
+	if got.State != StateIdle {
+		t.Fatalf("expected idle recovery view, got %q", got.State)
+	}
+	if got.ExactRecoveryRequires.AssistantPrefix {
+		t.Fatalf("expected idle thread not to require assistant-prefix continuation, got %#v", got.ExactRecoveryRequires)
+	}
+	if len(got.OutstandingToolCalls) != 2 {
+		t.Fatalf("expected two outstanding tool calls, got %#v", got.OutstandingToolCalls)
+	}
+
+	if got.OutstandingToolCalls[0].Call.CallID != "c1" ||
+		got.OutstandingToolCalls[0].Bound ||
+		got.OutstandingToolCalls[0].Started ||
+		got.OutstandingToolCalls[0].Continue != ToolContinueAuto ||
+		got.OutstandingToolCalls[0].Recovery != "" ||
+		len(got.OutstandingToolCalls[0].HandlerLoadData) != 0 {
+		t.Fatalf("unexpected requested outstanding tool call: %#v", got.OutstandingToolCalls[0])
+	}
+
+	if got.OutstandingToolCalls[1].Call.CallID != "c2" ||
+		!got.OutstandingToolCalls[1].Bound ||
+		!got.OutstandingToolCalls[1].Started ||
+		got.OutstandingToolCalls[1].Continue != ToolContinueManual ||
+		got.OutstandingToolCalls[1].Recovery != ToolRecoveryUnsafe ||
+		string(got.OutstandingToolCalls[1].HandlerLoadData) != `{"snapshot":"bound"}` {
+		t.Fatalf("unexpected started outstanding tool call: %#v", got.OutstandingToolCalls[1])
+	}
+
+	got.OutstandingToolCalls[1].HandlerLoadData[0] = 'X'
+	again := thread.RecoveryView()
+	if string(again.OutstandingToolCalls[1].HandlerLoadData) != `{"snapshot":"bound"}` {
+		t.Fatalf("expected cloned handler load data, got %#v", again.OutstandingToolCalls[1].HandlerLoadData)
+	}
+}
+
+func TestRecoveryViewConstructRequestDoesNotRequireAssistantPrefix(t *testing.T) {
+	thread := New()
+	thread.QueueItem(UserText("hello"))
+	thread.QueueItem(SendItem{})
+
+	got := thread.RecoveryView()
+	if got.State != StateConstructLLMRequest {
+		t.Fatalf("expected construct_llm_request, got %q", got.State)
+	}
+	if got.ExactRecoveryRequires.AssistantPrefix {
+		t.Fatalf("expected construct_llm_request not to require assistant-prefix continuation, got %#v", got.ExactRecoveryRequires)
+	}
+	if !got.CanRecoverExactlyWith(StreamerCapabilities{}) {
+		t.Fatalf("expected construct_llm_request exact recovery without assistant-prefix support, got %#v", got)
+	}
+}
+
+func TestRecoveryViewReceivingStreamRequiresAssistantPrefix(t *testing.T) {
+	thread := New()
+	streamStart := make(chan struct{})
+	thread.SetDelegate(ThreadDelegateFuncs{
+		OnRequest: func(_ *Thread) {
+			select {
+			case <-streamStart:
+			default:
+				close(streamStart)
+			}
+		},
+	})
+
+	streamer := newFakeStreamer().Reply(func(b *streamBuilder) {
+		b.Wait("hold")
+		b.Emit(AssistantText("world"))
+	})
+	thread.SetExecutor(NewThreadExecutor(streamer.Streamer()))
+
+	done := make(chan struct{})
+	go func() {
+		thread.QueueItem(UserText("hello"))
+		thread.QueueItem(SendItem{})
+		close(done)
+	}()
+
+	<-streamStart
+	cp, err := thread.Checkpoint(CheckpointOptions{Policy: InflightUnsafe})
+	if err != nil {
+		t.Fatalf("unsafe checkpoint: %v", err)
+	}
+	restored, err := RestoreCheckpoint(cp, RestoreOptions{AllowUnsafe: true})
+	if err != nil {
+		t.Fatalf("restore unsafe checkpoint: %v", err)
+	}
+
+	got := restored.RecoveryView()
+	if got.State != StateReceivingStream {
+		t.Fatalf("expected receiving_stream recovery view, got %q", got.State)
+	}
+	if !got.ExactRecoveryRequires.AssistantPrefix {
+		t.Fatalf("expected receiving_stream to require assistant-prefix continuation, got %#v", got.ExactRecoveryRequires)
+	}
+	if got.CanRecoverExactlyWith(StreamerCapabilities{}) {
+		t.Fatalf("expected exact recovery without assistant-prefix support to be rejected, got %#v", got)
+	}
+	if !got.CanRecoverExactlyWith(StreamerCapabilities{AssistantPrefix: true}) {
+		t.Fatalf("expected exact recovery with assistant-prefix support, got %#v", got)
+	}
+
+	streamer.Resolve("hold")
+	<-done
+}