Merge pull request #23 from beyond5959/dev

feat(codex): capture reasoning deltas in embedded runtime

Author: beyond5959

go.mod

@@ -3,7 +3,7 @@ module github.com/beyond5959/ngent
 go 1.24
 
 require (
-	github.com/beyond5959/acp-adapter v0.3.1-0.20260313104211-08e49ba1c367
+	github.com/beyond5959/acp-adapter v0.3.1
 	github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
 	modernc.org/sqlite v1.18.2
 )

go.sum

@@ -1,7 +1,5 @@
-github.com/beyond5959/acp-adapter v0.2.1-0.20260313090730-8c07c32144c3 h1:PUh2AuLRbTG3lgGRmrAIOBp9Juely/C4Ydp0GjheZVQ=
-github.com/beyond5959/acp-adapter v0.2.1-0.20260313090730-8c07c32144c3/go.mod h1:cr4I+9+la75oUge+Pr97KlcdQrkwIG6VwWHbc9ALxSE=
-github.com/beyond5959/acp-adapter v0.3.1-0.20260313104211-08e49ba1c367 h1:LOT0AfexFf14hiZlForYTkDbXBAWIu14nqXFP2Lya2Q=
-github.com/beyond5959/acp-adapter v0.3.1-0.20260313104211-08e49ba1c367/go.mod h1:cr4I+9+la75oUge+Pr97KlcdQrkwIG6VwWHbc9ALxSE=
+github.com/beyond5959/acp-adapter v0.3.1 h1:zxhs2cHWK9DT9sDUjAF8WonxPavIl0QK5rxh0NyxkYc=
+github.com/beyond5959/acp-adapter v0.3.1/go.mod h1:cr4I+9+la75oUge+Pr97KlcdQrkwIG6VwWHbc9ALxSE=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/google/go-cmp v0.5.3 h1:x95R7cp+rSeeqAMI2knLtQ0DKlaBhv2NrtrOvafPHRo=

internal/agents/codex/embedded.go

@@ -42,6 +42,7 @@ const (
 	stableSessionResolveRetries = 5
 	stableSessionResolveDelay   = 150 * time.Millisecond
 	initialSlashCommandsWait    = 250 * time.Millisecond
+	postPromptDrainTimeout      = 250 * time.Millisecond
 )
 
 // Config configures one embedded codex runtime provider instance.
@@ -467,14 +468,51 @@ func (c *Client) streamOnce(
 		promptDone <- promptResult{response: resp, err: reqErr}
 	}()
 
+	var (
+		finalStopReason agents.StopReason
+		promptFinished  bool
+		drainTimer      *time.Timer
+		drainCh         <-chan time.Time
+	)
+	stopDrainTimer := func() {
+		if drainTimer == nil {
+			return
+		}
+		if !drainTimer.Stop() {
+			select {
+			case <-drainTimer.C:
+			default:
+			}
+		}
+		drainCh = nil
+	}
+	resetDrainTimer := func() {
+		if drainTimer == nil {
+			drainTimer = time.NewTimer(postPromptDrainTimeout)
+			drainCh = drainTimer.C
+			return
+		}
+		if !drainTimer.Stop() {
+			select {
+			case <-drainTimer.C:
+			default:
+			}
+		}
+		drainTimer.Reset(postPromptDrainTimeout)
+		drainCh = drainTimer.C
+	}
+	defer stopDrainTimer()
+
 	for {
 		select {
 		case <-ctx.Done():
 			stopCancelWatcher()
+			stopDrainTimer()
 			return agents.StopReasonCancelled, nil
 		case result := <-promptDone:
-			stopCancelWatcher()
 			if result.err != nil {
+				stopCancelWatcher()
+				stopDrainTimer()
 				if errors.Is(result.err, context.Canceled) || errors.Is(result.err, context.DeadlineExceeded) || ctx.Err() != nil {
 					return agents.StopReasonCancelled, nil
 				}
@@ -483,15 +521,24 @@ func (c *Client) streamOnce(
 
 			stopReason, parseErr := parsePromptStopReason(result.response.Result)
 			if parseErr != nil {
+				stopCancelWatcher()
+				stopDrainTimer()
 				return agents.StopReasonEndTurn, parseErr
 			}
 			if stopReason == "cancelled" {
-				return agents.StopReasonCancelled, nil
+				finalStopReason = agents.StopReasonCancelled
+			} else {
+				finalStopReason = agents.StopReasonEndTurn
 			}
-			return agents.StopReasonEndTurn, nil
+			promptFinished = true
+			resetDrainTimer()
 		case msg, ok := <-updates:
 			if !ok {
 				stopCancelWatcher()
+				stopDrainTimer()
+				if promptFinished {
+					return finalStopReason, nil
+				}
 				if ctx.Err() != nil {
 					return agents.StopReasonCancelled, nil
 				}
@@ -500,8 +547,23 @@ func (c *Client) streamOnce(
 
 			if err := c.handleUpdate(ctx, runtime, msg, onDelta); err != nil {
 				stopCancelWatcher()
+				stopDrainTimer()
 				return agents.StopReasonEndTurn, err
 			}
+			if promptFinished {
+				if acpSessionUpdateIsTerminal(msg.Params) {
+					stopCancelWatcher()
+					stopDrainTimer()
+					return finalStopReason, nil
+				}
+				resetDrainTimer()
+			}
+		case <-drainCh:
+			stopCancelWatcher()
+			stopDrainTimer()
+			if promptFinished {
+				return finalStopReason, nil
+			}
 		}
 	}
 }
@@ -548,6 +610,7 @@ func (c *Client) handleUpdate(
 	observability.LogACPMessage(c.Name(), "inbound", msg)
 
 	if msg.Method == methodSessionUpdate {
+		updateType := acpSessionUpdateTopLevelType(msg.Params)
 		update, err := agents.ParseACPUpdate(msg.Params)
 		if err != nil {
 			return fmt.Errorf("codex: %w", err)
@@ -562,6 +625,15 @@ func (c *Client) handleUpdate(
 				return err
 			}
 			return nil
+		case agents.ACPUpdateTypeThoughtMessageChunk:
+			if updateType != "" && updateType != "reasoning" {
+				return nil
+			}
+			if err := agents.NotifyReasoningDelta(ctx, update.Delta); err != nil {
+				c.sendSessionCancel(runtime, c.currentSessionID())
+				return err
+			}
+			return nil
 		case agents.ACPUpdateTypePlan:
 			handler, ok := agents.PlanHandlerFromContext(ctx)
 			if !ok {
@@ -592,6 +664,41 @@ func (c *Client) handleUpdate(
 	return nil
 }
 
+func acpSessionUpdateTopLevelType(raw json.RawMessage) string {
+	if len(raw) == 0 {
+		return ""
+	}
+	var payload struct {
+		Type string `json:"type"`
+	}
+	if err := json.Unmarshal(raw, &payload); err != nil {
+		return ""
+	}
+	return strings.TrimSpace(payload.Type)
+}
+
+func acpSessionUpdateIsTerminal(raw json.RawMessage) bool {
+	if len(raw) == 0 {
+		return false
+	}
+	var payload struct {
+		Type   string `json:"type"`
+		Status string `json:"status"`
+	}
+	if err := json.Unmarshal(raw, &payload); err != nil {
+		return false
+	}
+	if strings.TrimSpace(payload.Type) != "status" {
+		return false
+	}
+	switch strings.TrimSpace(payload.Status) {
+	case "turn_completed", "turn_cancelled":
+		return true
+	default:
+		return false
+	}
+}
+
 func (c *Client) handlePermissionRequest(
 	ctx context.Context,
 	runtime *codexacp.EmbeddedRuntime,

internal/agents/codex/embedded_test.go

@@ -79,6 +79,69 @@ func TestStreamReplaysCachedSlashCommandsAfterConfigOptionsInit(t *testing.T) {
 	}
 }
 
+func TestStreamCapturesReasoningSummaryDeltas(t *testing.T) {
+	client := newFakeCodexClient(t)
+	defer func() {
+		_ = client.Close()
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	var reasoning strings.Builder
+	ctx = agents.WithReasoningHandler(ctx, func(_ context.Context, delta string) error {
+		reasoning.WriteString(delta)
+		return nil
+	})
+
+	var answer strings.Builder
+	stopReason, err := client.Stream(ctx, "reasoning summary probe", func(delta string) error {
+		answer.WriteString(delta)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("Stream(): %v", err)
+	}
+	if stopReason != agents.StopReasonEndTurn {
+		t.Fatalf("StopReason = %q, want %q", stopReason, agents.StopReasonEndTurn)
+	}
+
+	if got := answer.String(); !strings.Contains(got, "working") {
+		t.Fatalf("answer = %q, want it to include %q", got, "working")
+	}
+	if got, want := reasoning.String(), "Inspect repository state.\n\nConfirm reasoning plumbing."; got != want {
+		t.Fatalf("reasoning = %q, want %q", got, want)
+	}
+}
+
+func TestStreamCapturesReasoningTextDeltas(t *testing.T) {
+	client := newFakeCodexClient(t)
+	defer func() {
+		_ = client.Close()
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	var reasoning strings.Builder
+	ctx = agents.WithReasoningHandler(ctx, func(_ context.Context, delta string) error {
+		reasoning.WriteString(delta)
+		return nil
+	})
+
+	stopReason, err := client.Stream(ctx, "reasoning raw probe", func(string) error { return nil })
+	if err != nil {
+		t.Fatalf("Stream(): %v", err)
+	}
+	if stopReason != agents.StopReasonEndTurn {
+		t.Fatalf("StopReason = %q, want %q", stopReason, agents.StopReasonEndTurn)
+	}
+
+	if got, want := reasoning.String(), "Raw reasoning step 1. Raw reasoning step 2."; got != want {
+		t.Fatalf("reasoning = %q, want %q", got, want)
+	}
+}
+
 func TestSlashCommandsAfterConfigOptionsInit(t *testing.T) {
 	client := newFakeCodexClient(t)
 	defer func() {