oikos/cmd/nomos/continue.go

package main

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

	"github.com/google/uuid"
)

// execIDRe matches "execution <uuid>" in a tool result — the phrasing shared
// by request_execution / run when they queue or start a gated execution.
// Only these async executions need continuation; the synchronous auto-run
// path returns its output inline and is already observed in-turn.
var execIDRe = regexp.MustCompile(`(?i)execution\s+([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})`)

func extractExecutionIDs(toolResult string) []uuid.UUID {
	matches := execIDRe.FindAllStringSubmatch(toolResult, -1)
	seen := map[uuid.UUID]bool{}
	var out []uuid.UUID
	for _, m := range matches {
		if id, err := uuid.Parse(m[1]); err == nil && !seen[id] {
			seen[id] = true
			out = append(out, id)
		}
	}
	return out
}

// runContinuationWorker is the event loop that replaces the human typing
// "continue". It polls for gated executions that (a) were initiated by a chat
// session and (b) have just finished, and — while that agent has an open assent
// window (an approved plan is in flight) — feeds each result back into the
// agent so it proceeds to the next step or recovers from the failure, all
// without an operator tick. Blocks until ctx is cancelled.
func (a *agent) runContinuationWorker(ctx context.Context) {
	if a.store == nil {
		slog.Warn("nomos: continuation worker disabled (no store)")
		return
	}
	slog.Info("nomos: continuation worker started")
	ticker := time.NewTicker(4 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			a.processContinuations(ctx)
		}
	}
}

func (a *agent) processContinuations(ctx context.Context) {
	pending := a.store.pendingContinuations(ctx, 5)
	for _, p := range pending {
		// Scope gate: only auto-continue while an approved plan is active FOR
		// THIS SESSION. Checked per-item, not once for the whole batch — with
		// multiple tasks in flight, one task's open window must never cover a
		// pending continuation belonging to a different task.
		if !a.store.assentWindowActive(ctx, a.agentID, p.SessionID) {
			// A finished one-off execution with no window is left as-is
			// (marked continued so we don't re-check it forever) — the
			// operator decides what happens next, as today.
			a.store.markContinued(ctx, p.ExecID)
			continue
		}
		a.store.markContinued(ctx, p.ExecID) // stamp first: a failure here must not cause a re-continue loop
		a.continueSession(ctx, p)
	}
}

// continueSession re-invokes the agent for one finished execution. Persists
// progress LIVE — a placeholder row immediately, updated in place as each
// tool call completes — instead of only saving once the whole continuation
// finishes. The frontend polls (see chat.ts startPolling); without
// incremental persistence here, a continuation that runs several tool calls
// before concluding would look like total silence in the UI for however long
// that takes, which is exactly the "I just wait while nothing happens"
// complaint this exists to fix — polling alone only helps if there's
// something new to poll for.
func (a *agent) continueSession(ctx context.Context, p pendingContinuation) {
	slog.Info("nomos: auto-continuing session", "session", p.SessionID, "execution", p.ExecID, "status", p.Status)
	a.resumeSession(ctx, p.SessionID, buildContinuationNote(p))
}

// resumeSession re-invokes the agent for a session with a system-injected note —
// a finished execution (continueSession) or an operator's answer to a question
// (handleAnswerQuestion) — persisting progress LIVE (a placeholder row updated
// in place as each tool call lands) so the frontend poller sees each step,
// instead of total silence until the whole resume concludes.
func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
	placeholder, _ := json.Marshal(map[string]any{
		"role": "assistant",
		"text": "",
		"auto": true,
	})
	msgID, err := a.store.insertMessageReturningID(ctx, sessionID, "assistant", placeholder)
	if err != nil {
		slog.Error("nomos: resume placeholder insert failed", "session", sessionID, "error", err)
	}

	var toolCalls []map[string]any
	var finalText, errText string

	persist := func() {
		if msgID == uuid.Nil {
			return
		}
		text := finalText
		if text == "" && errText != "" {
			text = fmt.Sprintf("(auto-continuation hit an internal error and did not respond: %s — the execution's own result is above; you may need to prompt the agent again)", errText)
		}
		body, _ := json.Marshal(map[string]any{
			"role":       "assistant",
			"text":       text,
			"tool_calls": toolCalls,
			"auto":       true, // marks this as an autonomous continuation, not an operator turn
		})
		a.store.updateMessage(ctx, msgID, body)
	}

	// One retry if the LLM call itself produced nothing (transient flake /
	// empty-response) — the whole point of this mechanism is "don't give up
	// on the first error," which should apply to the continuation call
	// itself, not just the homelab commands it's continuing. Found live: a
	// destructive-recovery continuation hit an empty LLM response, its
	// internal retry (chatWith's own maxLLMRetries=1) also came up empty, and
	// without this outer retry the operator would see nothing at all.
	cctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
	defer cancel()
	for attempt := 0; attempt < 2; attempt++ {
		toolCalls, finalText, errText = nil, "", ""
		emit := 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
					toolCalls = append(toolCalls, m)
				}
				persist() // live: a poller sees this step land within seconds
			}
			if ev.Type == "text" {
				finalText, _ = ev.Data.(string)
			}
			if ev.Type == "error" {
				errText, _ = ev.Data.(string)
			}
		}
		a.chatWith(cctx, sessionID, "", note, emit)
		if finalText != "" || len(toolCalls) > 0 {
			break
		}
		if attempt == 0 {
			slog.Warn("nomos: resume produced nothing, retrying once", "session", sessionID, "error", errText)
		}
	}

	if errText != "" && finalText == "" {
		slog.Error("nomos: resume produced no response after retry", "session", sessionID, "error", errText)
	}
	persist() // final state — same row, updated one last time with the concluding text
}

// buildContinuationNote frames the finished execution for the model: what
// happened, and what to do about it. The persist-through-errors instruction
// lives here (and in SOUL) so the agent recovers instead of stopping.
func buildContinuationNote(p pendingContinuation) string {
	action := p.Action
	if i := strings.IndexByte(action, ':'); i > 0 && len(action) > 40 {
		action = action[:i] // keep just the action verb for brevity; params are in the DB
	}
	result := p.Result
	if len(result) > 3000 {
		result = result[:3000] + "…[truncated]"
	}
	var b strings.Builder
	fmt.Fprintf(&b, "[System: execution %s (%s) finished with status=%s.\nResult: %s\n\n",
		p.ExecID, action, p.Status, result)
	switch p.Status {
	case "completed":
		b.WriteString("It SUCCEEDED. Continue the approved plan: run the next step. If this was the final step, verify the end goal actually works (e.g. curl the service) and then report success to the operator. Do NOT stop and wait for the operator to say 'continue'.")
	case "failed", "cancelled":
		b.WriteString("It FAILED. Do NOT give up or hand back to the operator. Diagnose the cause from the result above (and by running read-only inspection commands if needed), form a hypothesis, fix it, and retry or take an alternative approach. You have an active assent window, so config_mutation steps run without re-approval. Only stop and ask the operator if you are genuinely blocked (need information only they have) or the fix would require a destructive action they haven't approved.")
	default: // denied / revoked
		b.WriteString("The operator denied or revoked this step. Stop executing this plan and briefly acknowledge.")
	}
	b.WriteString("]")
	return b.String()
}