oikos/cmd/nomos/workers.go

package main

import (
	"context"
	"encoding/json"
	"log/slog"
	"strings"
	"time"

	"github.com/google/uuid"
)

// runChatTurn is the shared core of an operator-initiated turn: insert an
// assistant placeholder, run a.chat with incremental persistence (so whatever
// happened before an abort is never lost), finalize the row, and derive a
// title. It is agnostic to the transport: `sink` receives every agent event
// for delivery (SSE for a live handleChat, a no-op for a queued turn that has
// no client attached — the frontend learns about those via the poller + the
// status-driven "working" signal). The caller MUST already hold the session's
// turn-gate permit.
func (a *agent) runChatTurn(pctx, ctx context.Context, sessionID, message string, sink func(agentEvent)) {
	toolCalls := []map[string]any{}
	// P3: accumulate per-iteration reasoning instead of overwriting with the
	// final `text` event (see the original inline comment in handleChat).
	var textParts []string
	var thinkingParts []string
	var finalText string
	var finalThinking string

	placeholder, _ := json.Marshal(map[string]any{"role": "assistant", "text": ""})
	msgID, err := a.store.insertMessageReturningID(pctx, sessionID, "assistant", placeholder)
	if err != nil {
		slog.Error("nomos: chat placeholder insert failed", "session", sessionID, "error", err)
	}
	persist := func() {
		if msgID == uuid.Nil {
			return
		}
		body, _ := json.Marshal(map[string]any{
			"role":       "assistant",
			"text":       finalText,
			"thinking":   finalThinking,
			"tool_calls": toolCalls,
		})
		a.store.updateMessage(pctx, msgID, body)
	}

	a.chat(ctx, sessionID, message, func(ev agentEvent) {
		if ev.Type == "tool_use" || ev.Type == "tool_result" {
			if m, ok := ev.Data.(map[string]any); ok {
				m["type"] = ev.Type
				// One entry per tool call: tool_use creates it, tool_result
				// merges the result into the same entry (matched by id).
				id, _ := m["id"].(string)
				if id != "" && ev.Type == "tool_result" {
					for _, existing := range toolCalls {
						if eID, _ := existing["id"].(string); eID == id {
							for k, v := range m {
								existing[k] = v
							}
							break
						}
					}
				} else {
					toolCalls = append(toolCalls, m)
				}
			}
			persist() // live: survives even if the client disconnects right after
		}
		if ev.Type == "text" {
			if t, ok := ev.Data.(string); ok && t != "" {
				if ev.IsThinking {
					thinkingParts = append(thinkingParts, t)
					finalThinking = strings.Join(thinkingParts, "\n\n")
				} else {
					textParts = append(textParts, t)
					finalText = strings.Join(textParts, "\n\n")
				}
				persist()
			}
		}
		sink(ev)
	})

	// B.6: if the turn ended with no text and no tool calls (the model
	// empty-response'd and all retries failed), delete the placeholder row
	// instead of persisting an empty bubble.
	if finalText == "" && len(toolCalls) == 0 && msgID != uuid.Nil {
		a.store.deleteMessage(pctx, msgID)
	} else {
		persist() // final state — same row, updated one last time
	}

	// Title: prefer the goal once set; else the first assistant answer.
	if finalText != "" && sessionID != "ephemeral" {
		var goalTitle string
		if sess, gerr := a.store.getSession(pctx, sessionID); gerr == nil && sess.Goal != "" {
			goalTitle = truncate(sess.Goal, 120)
		}
		title := goalTitle
		if title == "" {
			title = truncate(finalText, 80)
		}
		if title != "" {
			a.store.updateSessionTitle(pctx, sessionID, title)
		}
	}
}

// drainAcquireWait is how long drainQueued blocks for a busy gate before
// re-queuing and deferring to the holder's own release-drain. A package var so
// tests can shorten it; in production it just needs to outlast the brief
// release→drain handoff window.
var drainAcquireWait = 5 * time.Second

// drainQueued runs every queued operator message for a session as its own turn,
// one at a time, under the turn gate. Called (in a goroutine) whenever a turn
// releases the gate — from handleChat (live) and resumeSession (background) —
// so a message queued while the agent was busy is acted on as soon as it's
// free, without the operator re-sending. See messagequeue.go (plan 2026-08-03
// F2).
//
// Each queued turn is persisted incrementally and has no SSE client (the
// browser detached after receiving the `queued` event); the frontend sees the
// result via the 3s poller and the status-driven "working" indicator.
func (a *agent) drainQueued(ctx context.Context, sessionID string) {
	for {
		msg, ok := a.queue.dequeue(sessionID)
		if !ok {
			return
		}
		// Block briefly for the gate. If a live turn grabbed it first, put the
		// message back — that turn's release will drain it again. Never stack.
		if !a.gate.acquire(sessionID, drainAcquireWait) {
			a.queue.requeueFront(sessionID, msg)
			return
		}
		slog.Info("nomos: running queued operator message", "session", sessionID)
		pctx := context.Background()
		// Run the turn inside a per-iteration closure so the gate release is
		// deferred to the end of THIS turn (and runs even if runChatTurn
		// panics — safego recovers the panic at the goroutine boundary, so a
		// non-deferred release would be skipped and the session's permit held
		// forever, deadlocking all future turns). A bare `defer release` in
		// the loop would be wrong too: Go defers run at function exit, not
		// iteration exit, so the gate would stay held across iterations.
		func() {
			defer a.gate.release(sessionID)
			a.runChatTurn(pctx, ctx, sessionID, msg, func(agentEvent) {})
		}()
	}
}