diff --git a/threads/EXECUTOR_RECOVERY.md b/threads/EXECUTOR_RECOVERY.md index 0fd0669..33bca26 100644 --- a/threads/EXECUTOR_RECOVERY.md +++ b/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 diff --git a/threads/README.md b/threads/README.md index dc1129f..cdd3b13 100644 --- a/threads/README.md +++ b/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 diff --git a/threads/controlblock.go b/threads/controlblock.go index 5796148..07a98e0 100644 --- a/threads/controlblock.go +++ b/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: diff --git a/threads/recovery_view.go b/threads/recovery_view.go new file mode 100644 index 0000000..ff5a391 --- /dev/null +++ b/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{} +} diff --git a/threads/recovery_view_test.go b/threads/recovery_view_test.go new file mode 100644 index 0000000..7626af0 --- /dev/null +++ b/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 +}