Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion apps/daemon/internal/agent/claudesdk/execution_controls_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ func TestExecutionControlsRejectUnsupportedProfilesBeforeLaunch(t *testing.T) {
"unknown-verbosity": {TextVerbosity: "invalid"},
"no-effort": {ReasoningEffort: "none"},
"minimal-effort": {ReasoningEffort: "minimal"},
"summary": {ReasoningSummary: "auto"},
"summary": {ReasoningSummary: "concise"},
"service-tier": {ServiceTier: "priority"},
}
for name, controls := range cases {
Expand Down Expand Up @@ -169,3 +169,17 @@ func TestToolDiscoveryPreservesFrozenFunctionsBesideOtherProfiles(t *testing.T)
t.Fatal("deferred definitions became eager")
}
}

func TestReasoningSummaryReachesTheBridgeOnlyWhenRequested(t *testing.T) {
root := t.TempDir()
t.Setenv("OAC_RUNTIME_HOME", root)
config := Config{Entrypoint: filepath.Join(root, "worker"), StateDir: filepath.Join(root, "state")}
request := proto.PromptRequestPayload{RunID: "run", Input: proto.TextInput("Input."), AgentOptions: map[string]any{"model": "native-model"}}
for _, summary := range []string{"", "auto"} {
request.ExecutionControls = &proto.ExecutionControls{TextVerbosity: "medium", ReasoningSummary: summary}
start, _, err := prepare(config, request)
if err != nil || start.ReasoningSummary != (summary != "") {
t.Fatalf("summary %q: reasoning flag %v, err %v", summary, start.ReasoningSummary, err)
}
}
}
8 changes: 8 additions & 0 deletions apps/daemon/internal/agent/claudesdk/executor_turn.go
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,14 @@ func (s *session) runTurn(start startRequest, out chan<- proto.Envelope) {
break
}
emit(proto.TypeOutputMessage, message)
case "reasoning":
reasoning := event.Reasoning
if !start.ReasoningSummary || reasoning == nil || !reasoning.Valid() {
failure = fmt.Errorf("claudesdk: invalid reasoning observation")
s.invalidate()
break
}
emit(proto.TypeReasoning, reasoning)
case "function_call", "function_withdrawn", "function_applied":
if err := s.receiveFunction(event, start, emit); err != nil {
failure = err
Expand Down
2 changes: 2 additions & 0 deletions apps/daemon/internal/agent/claudesdk/options.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ type startRequest struct {
NativeModelOptions *nativeModelOptions `json:"native_model_options,omitempty"`
ToolSearch bool `json:"tool_search,omitempty"`
WebSearch string `json:"web_search,omitempty"`
ReasoningSummary bool `json:"reasoning_summary,omitempty"`
Subagents *subagentOptions `json:"subagents,omitempty"`
OutputFormat *proto.OutputFormat `json:"output_format,omitempty"`
Type string `json:"type"`
Expand Down Expand Up @@ -80,6 +81,7 @@ func prepareConfiguration(config Config, req proto.PromptRequestPayload) (startR
if controls := req.ExecutionControls; controls != nil && controls.WebSearch != nil {
start.WebSearch = controls.WebSearch.Mode
}
start.ReasoningSummary = req.ExecutionControls != nil && req.ExecutionControls.ReasoningSummary != ""
if req.ExecutionControls != nil && req.ExecutionControls.OutputFormat != nil {
format := req.ExecutionControls.OutputFormat
if format.Type != "json_schema" {
Expand Down
1 change: 1 addition & 0 deletions apps/daemon/internal/agent/claudesdk/session.go
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ type bridgeEvent struct {
Code string `json:"code"`
ItemID string `json:"item_id"`
Message *proto.OutputMessagePayload `json:"message"`
Reasoning *proto.ReasoningPayload `json:"reasoning"`
Call *proto.FunctionCallPayload `json:"call"`
Withdrawn *proto.FunctionCallWithdrawnPayload `json:"withdrawn"`
CallID string `json:"call_id"`
Expand Down
13 changes: 10 additions & 3 deletions apps/daemon/internal/agent/codex/execution_controls_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@ func TestExecutionControlsMapToNativeSettings(t *testing.T) {
controls *proto.ExecutionControls
want [][2]string
}{
{"absent controls send nothing", nil, nil},
{"empty controls send nothing", &proto.ExecutionControls{}, nil},
{"absent controls turn summaries off", nil, nil},
{"empty controls turn summaries off", &proto.ExecutionControls{}, nil},
{"verbosity", &proto.ExecutionControls{TextVerbosity: "high"}, [][2]string{{"model_verbosity", `"high"`}}},
{"search without settings", &proto.ExecutionControls{WebSearch: &proto.WebSearchControl{Mode: "live", ContextSize: "medium"}},
[][2]string{{"web_search", `"live"`}, {"tools.web_search.context_size", `"medium"`}}},
Expand All @@ -37,9 +37,16 @@ func TestExecutionControlsMapToNativeSettings(t *testing.T) {
t.Fatal(err)
}
plan.Cleanup()
if want := append(append([][2]string{}, base...), tc.want...); !reflect.DeepEqual(plan.ExtraConfig, want) {
want := append([][2]string{}, base...)
if tc.controls == nil || tc.controls.ReasoningSummary == "" {
want = append(want, [2]string{"model_reasoning_summary", `"none"`})
}
if want = append(want, tc.want...); !reflect.DeepEqual(plan.ExtraConfig, want) {
t.Fatalf("config = %v, want %v", plan.ExtraConfig, want)
}
if plan.ReasoningSummary != (tc.controls != nil && tc.controls.ReasoningSummary != "") {
t.Fatalf("reasoning summary gate = %v", plan.ReasoningSummary)
}
})
}
}
Expand Down
2 changes: 1 addition & 1 deletion apps/daemon/internal/agent/codex/executor.go
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ func (e *Executor) StartTurn(ctx context.Context, runID string, input proto.Mess
functions: functions, observeSubagentIdentities: base.observeSubagentIdentities,
cfg: base.cfg, rpc: base.rpc, cancelCtx: turnCtx, cancelFn: cancel,
waitDone: make(chan struct{}), outputDone: make(chan struct{}), cleanup: func() {},
bufs: NewItemBuffers(), resolvedModel: base.resolvedModel, runID: runID, out: out}
bufs: NewItemBuffers(), resolvedModel: base.resolvedModel, reasoningSummary: base.reasoningSummary, runID: runID, out: out}
if previous != nil {
s.threadID = previous.currentThreadID()
s.retiredTurns = make(map[string]bool, len(previous.retiredTurns)+1)
Expand Down
10 changes: 9 additions & 1 deletion apps/daemon/internal/agent/codex/options.go
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ type SessionPlan struct {
// ModelReasoningEffort is frozen for launch and every native Turn.
ModelReasoningEffort string

// ReasoningSummary reports that the Agent requested reasoning summaries.
ReasoningSummary bool

// OutputSchema is the frozen public JSON Schema sent as outputSchema on every
// native root Turn.
OutputSchema json.RawMessage
Expand Down Expand Up @@ -160,10 +163,14 @@ func excludeFromShellEnvironment(plan *SessionPlan, pattern string) {
// absent control sends nothing, so Codex keeps its native default; Codex itself
// ignores a value the selected model does not support.
func applyExecutionControls(plan *SessionPlan, controls *proto.ExecutionControls) {
set := func(key, value string) { plan.ExtraConfig = append(plan.ExtraConfig, [2]string{key, value}) }
if controls == nil || controls.ReasoningSummary == "" {
// The official default is no reasoning summary, while Codex defaults to a model-dependent one.
set("model_reasoning_summary", strconv("none"))
}
if controls == nil {
return
}
set := func(key, value string) { plan.ExtraConfig = append(plan.ExtraConfig, [2]string{key, value}) }
if search := controls.WebSearch; search != nil {
set("web_search", strconv(search.Mode))
set("tools.web_search.context_size", strconv(search.ContextSize))
Expand Down Expand Up @@ -193,6 +200,7 @@ func applyExecutionControls(plan *SessionPlan, controls *proto.ExecutionControls
set("model_reasoning_effort", strconv(controls.ReasoningEffort))
}
if controls.ReasoningSummary != "" {
plan.ReasoningSummary = true
set("model_reasoning_summary", strconv(controls.ReasoningSummary))
}
if controls.ServiceTier != "" {
Expand Down
1 change: 1 addition & 0 deletions apps/daemon/internal/agent/codex/preparation.go
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ func newPreparation(parent context.Context, req proto.PromptRequestPayload, cfg
cleanup: sync.OnceFunc(plan.Cleanup),
bufs: NewItemBuffers(),
resolvedModel: plan.Model,
reasoningSummary: plan.ReasoningSummary,
}
plan.Cleanup = s.cleanup
p := &Prepared{
Expand Down
16 changes: 10 additions & 6 deletions apps/daemon/internal/agent/codex/protocol.go
Original file line number Diff line number Diff line change
Expand Up @@ -216,11 +216,10 @@ type ThreadItem struct {
ID string `json:"id,omitempty"`

// agentMessage / reasoning
Phase string `json:"phase,omitempty"`
Text string `json:"text,omitempty"`
Summary []string `json:"summary,omitempty"`
Content []string `json:"content,omitempty"`
SummaryText string `json:"summary_text,omitempty"`
Phase string `json:"phase,omitempty"`
Text string `json:"text,omitempty"`
Summary []string `json:"summary,omitempty"`
Content []string `json:"content,omitempty"`

// commandExecution
Command string `json:"command,omitempty"`
Expand Down Expand Up @@ -283,7 +282,12 @@ type AgentMessageDeltaNotification struct {
Delta string `json:"delta"`
}

type ReasoningDeltaNotification = AgentMessageDeltaNotification
// ReasoningDeltaNotification is a summaryTextDelta (SummaryIndex) or a textDelta (ContentIndex).
type ReasoningDeltaNotification struct {
AgentMessageDeltaNotification
SummaryIndex int `json:"summaryIndex"`
ContentIndex int `json:"contentIndex"`
}

type ThreadTokenUsageUpdatedNotification struct {
ThreadID string `json:"threadId"`
Expand Down
65 changes: 53 additions & 12 deletions apps/daemon/internal/agent/codex/session.go
Original file line number Diff line number Diff line change
Expand Up @@ -86,8 +86,7 @@ type Session struct {
threadID string
steering steeringTurn

deltaSeq atomic.Uint64
thinkingSeq atomic.Uint64
deltaSeq atomic.Uint64

bufs *ItemBuffers

Expand All @@ -99,6 +98,8 @@ type Session struct {
resumeUsageThreadID string
resumeUsageTotal *TurnUsage
resolvedModel string
// reasoningSummary reports that the Agent requested reasoning summaries.
reasoningSummary bool

finalTextMu sync.Mutex
finalText string
Expand All @@ -125,8 +126,8 @@ func (s *Session) registerHandlers() {
rpc.OnNotification("item/completed", s.onItemCompleted)
rpc.OnNotification("item/agentMessage/delta", s.onAgentDelta)
rpc.OnNotification("item/commandExecution/outputDelta", s.onCommandOutput)
rpc.OnNotification("item/reasoning/textDelta", s.onReasoningDelta)
rpc.OnNotification("item/reasoning/summaryTextDelta", s.onReasoningDelta)
rpc.OnNotification("item/reasoning/textDelta", func(raw json.RawMessage) { s.onReasoningDelta("text", raw) })
rpc.OnNotification("item/reasoning/summaryTextDelta", func(raw json.RawMessage) { s.onReasoningDelta("summary", raw) })
rpc.OnNotification("thread/tokenUsage/updated", s.onUsageUpdated)
rpc.OnNotification("error", s.onErrorNotif)

Expand All @@ -143,21 +144,61 @@ func (s *Session) onTurnStarted(raw json.RawMessage) {
}
}

func (s *Session) onReasoningDelta(raw json.RawMessage) {
var p AgentMessageDeltaNotification
// onReasoningDelta streams the first native channel a reasoning item delivers:
// its summary, or the raw reasoning text of a model that has no summary. Each
// native part is reported as its own part.
func (s *Session) onReasoningDelta(channel string, raw json.RawMessage) {
var p ReasoningDeltaNotification
if err := json.Unmarshal(raw, &p); err != nil {
return
}
if !s.isRootTurn(p.ThreadID, p.TurnID) || p.Delta == "" || p.ItemID == "" {
if !s.reasoningSummary || !s.isRootTurn(p.ThreadID, p.TurnID) || p.Delta == "" || p.ItemID == "" {
return
}
_ = FoldDeltaIntoBuffer(s.bufs, "reasoning", p.ItemID, p.Delta)
seq := s.thinkingSeq.Add(1)
env, err := proto.NewEnvelope(proto.TypeThinking, s.runID, proto.ThinkingPayload{Text: p.Delta, Sequence: seq})
if err != nil {
native := p.ContentIndex
if channel == "summary" {
native = p.SummaryIndex
}
stream, started := s.bufs.Reasoning[p.ItemID]
if started && stream.channel != channel {
return
}
s.trySend(env)
if started && stream.native != native {
stream.part++
}
stream.channel, stream.native = channel, native
s.bufs.Reasoning[p.ItemID] = stream
s.sendReasoning(p.ItemID, "in_progress", stream.part, p.Delta)
}

// completeReasoning ends a reasoning item. An item that streamed nothing is
// reported whole from its completed body; one with no text is not reported.
func (s *Session) completeReasoning(item ThreadItem) {
if !s.reasoningSummary {
return
}
_, streamed := s.bufs.Reasoning[item.ID]
delete(s.bufs.Reasoning, item.ID)
if !streamed {
part := 0
for _, text := range nativeReasoningParts(item) {
if text != "" {
s.sendReasoning(item.ID, "in_progress", part, text)
part++
}
}
if part == 0 {
return
}
}
s.sendReasoning(item.ID, "completed", 0, "")
}

func (s *Session) sendReasoning(id, status string, part int, delta string) {
env, err := proto.NewEnvelope(proto.TypeReasoning, s.runID, proto.ReasoningPayload{ID: id, Status: status, Index: part, Delta: delta})
if err == nil {
s.trySend(env)
}
}

func (s *Session) onTurnCompleted(raw json.RawMessage) {
Expand Down
Loading
Loading