oikos/cmd/nomos/eval/main.go

// Command nomos-eval runs golden conversation evals against a live nomos
// gateway. It loads a YAML manifest of conversations + assertions, sends
// each prompt to the chat endpoint, waits for the turn(s) to finish, and
// scores assertions against the persisted transcript.
//
// Usage:
//
//	go run ./cmd/nomos/eval -gateway http://localhost:8092 -manifest evals/*.yaml
//
// The gateway must already be running (nomos serve, or the docker container).
// Each conversation costs real OpenRouter credits (~$0.01–0.05 each).
//
// Manifest format — see evals/example.yaml. Assertions are scored against the
// final transcript: tool calls made, plan steps, final session status, and
// whether the turn completed. The runner does NOT judge text quality — only
// structural properties that can be checked deterministically from the
// persisted state. This is deliberate: text quality is model-dependent and
// noisy; structure is what the Go gates + SOUL.md should enforce.
package main

import (
	"bytes"
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"time"
)

func main() {
	gateway := flag.String("gateway", "http://localhost:8092", "nomos gateway URL")
	manifestGlob := flag.String("manifest", "evals/*.yaml", "glob of manifest files to run")
	timeout := flag.Duration("timeout", 2*time.Minute, "per-conversation timeout")
	flag.Parse()

	if err := health(*gateway); err != nil {
		fmt.Fprintf(os.Stderr, "gateway not reachable at %s: %v\n", *gateway, err)
		os.Exit(1)
	}

	files, err := filepath.Glob(*manifestGlob)
	if err != nil {
		fmt.Fprintf(os.Stderr, "glob %s: %v\n", *manifestGlob, err)
		os.Exit(1)
	}
	if len(files) == 0 {
		fmt.Fprintf(os.Stderr, "no manifests matched %s\n", *manifestGlob)
		os.Exit(1)
	}

	total, passed, failed := 0, 0, 0
	for _, f := range files {
		convs, err := loadManifest(f)
		if err != nil {
			fmt.Fprintf(os.Stderr, "load %s: %v\n", f, err)
			os.Exit(1)
		}
		for _, c := range convs {
			total++
			name := c.Name
			if name == "" {
				name = fmt.Sprintf("conversation-%d", total)
			}
			fmt.Printf("=== %s (from %s) ===\n", name, filepath.Base(f))
			res := runConversation(context.Background(), *gateway, c, *timeout)
			if res.Passed {
				passed++
				fmt.Printf("  ✅ PASS (%.1fs, %d tool calls)\n", res.Duration.Seconds(), res.ToolCallCount)
			} else {
				failed++
				fmt.Printf("  ❌ FAIL (%.1fs, %d tool calls)\n", res.Duration.Seconds(), res.ToolCallCount)
			}
			for _, a := range res.Assertions {
				mark := "✅"
				if !a.Passed {
					mark = "❌"
				}
				fmt.Printf("    %s %s: %s\n", mark, a.Name, a.Detail)
			}
		}
	}
	fmt.Printf("\n=== Summary: %d/%d passed, %d failed ===\n", passed, total, failed)
	if failed > 0 {
		os.Exit(1)
	}
}

func health(gateway string) error {
	resp, err := http.Get(gateway + "/healthz")
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return fmt.Errorf("healthz status %d", resp.StatusCode)
	}
	return nil
}

// runConversation sends the prompt (and any followup), waits for each turn to
// finish, then scores assertions against the final transcript.
func runConversation(ctx context.Context, gateway string, c conversation, timeout time.Duration) convResult {
	start := time.Now()
	deadline := time.Now().Add(timeout)
	res := convResult{}

	// Send the initial prompt (no session_id → creates a new session).
	sid, err := sendChat(ctx, gateway, "", c.Prompt)
	if err != nil {
		res.Assertions = []assertionResult{{Name: "send_prompt", Passed: false, Detail: err.Error()}}
		res.Duration = time.Since(start)
		return res
	}
	res.SessionID = sid

	// Wait for the first turn to finish.
	if err := waitForTurn(ctx, gateway, sid, deadline); err != nil {
		res.Assertions = []assertionResult{{Name: "turn_complete", Passed: false, Detail: err.Error()}}
		res.Duration = time.Since(start)
		return res
	}

	// Send followup if any.
	for _, fu := range c.followups() {
		if _, err := sendChat(ctx, gateway, sid, fu); err != nil {
			res.Assertions = []assertionResult{{Name: "send_followup", Passed: false, Detail: err.Error()}}
			res.Duration = time.Since(start)
			return res
		}
		if err := waitForTurn(ctx, gateway, sid, deadline); err != nil {
			res.Assertions = []assertionResult{{Name: "followup_turn_complete", Passed: false, Detail: err.Error()}}
			res.Duration = time.Since(start)
			return res
		}
	}

	// Fetch the final transcript + session state.
	transcript, session, err := fetchTranscript(ctx, gateway, sid)
	if err != nil {
		res.Assertions = []assertionResult{{Name: "fetch_transcript", Passed: false, Detail: err.Error()}}
		res.Duration = time.Since(start)
		return res
	}
	res.ToolCallCount = transcript.toolCallCount()
	res.Duration = time.Since(start)

	// Score assertions.
	res.Assertions = scoreAssertions(c.Assertions, transcript, session)

	res.Passed = true
	for _, a := range res.Assertions {
		if !a.Passed {
			res.Passed = false
			break
		}
	}
	return res
}

// sendChat POSTs to /chat and extracts the session_id from the first SSE
// event, then KEEPS READING the stream until it ends (the `done` event or
// the connection closes). This is critical: the chat handler uses
// r.Context() which cancels when the HTTP connection closes — if we stop
// reading after the session event, the agent's work gets canceled mid-turn.
// We must drain the full stream so the agent completes its turn server-side.
func sendChat(ctx context.Context, gateway, sid, message string) (string, error) {
	body, _ := json.Marshal(map[string]string{"session_id": sid, "message": message})
	req, _ := http.NewRequestWithContext(ctx, "POST", gateway+"/chat", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 && resp.StatusCode != 202 {
		b, _ := io.ReadAll(resp.Body)
		return "", fmt.Errorf("chat status %d: %s", resp.StatusCode, string(b))
	}
	// For a reconnect (sid != ""), the body is 202 with no stream.
	if sid != "" {
		io.Copy(io.Discard, resp.Body)
		return sid, nil
	}
	// Read the SSE stream, capturing the session_id from the first session
	// event, and draining the rest so the agent's turn completes. The stream
	// ends when the server closes it (after the `done` event) or when the
	// request context cancels.
	dec := newSSEReader(resp.Body)
	sessionID := ""
	for {
		ev, err := dec.next()
		if err != nil {
			if sessionID == "" {
				return "", fmt.Errorf("no session event before stream end: %w", err)
			}
			return sessionID, nil
		}
		if ev["type"] == "session" && sessionID == "" {
			if s, ok := ev["session_id"].(string); ok {
				sessionID = s
			}
		}
		// Keep reading until the stream ends — don't return early.
	}
}

// waitForTurn polls the session until its last_active_at stops advancing for
// 8 seconds (the turn ended) or the session reaches a terminal status. We
// can't rely on status=done alone because a trivial task may auto-complete
// while a plan-proposing task stays in 'executing' waiting for approval.
func waitForTurn(ctx context.Context, gateway, sid string, deadline time.Time) error {
	var lastActive string
	stableSince := time.Now()
	for {
		if time.Now().After(deadline) {
			return fmt.Errorf("timeout waiting for turn to complete")
		}
		_, session, err := fetchTranscript(ctx, gateway, sid)
		if err != nil {
			time.Sleep(2 * time.Second)
			continue
		}
		if session.LastActive != lastActive {
			lastActive = session.LastActive
			stableSince = time.Now()
		}
		if time.Since(stableSince) >= 8*time.Second {
			return nil // turn is idle — consider it complete
		}
		if session.Status == "done" || session.Status == "failed" {
			return nil
		}
		time.Sleep(2 * time.Second)
	}
}

type transcript struct {
	Messages []struct {
		Role    string `json:"role"`
		Content struct {
			Text      string            `json:"text"`
			ToolCalls []map[string]any  `json:"tool_calls"`
		} `json:"content"`
	} `json:"messages"`
	// PlanSteps is fetched from /sessions/{id}/plan (P5 plan_generations
	// assertion). Each step carries a `generation` int; distinctGenerations
	// counts the unique values. nil when the endpoint returned no plan
	// (e.g. a pure-DB Q&A with no propose_plan call).
	PlanSteps []planStep `json:"steps"`
}

// planStep is one step from /sessions/{id}/plan, carrying only the fields the
// eval needs: the generation number (P2 iteration counter).
type planStep struct {
	Generation int    `json:"generation"`
	Status     string `json:"status"`
	Title      string `json:"title"`
}

func (t transcript) toolCallCount() int {
	n := 0
	for _, m := range t.Messages {
		n += len(m.Content.ToolCalls)
	}
	return n
}

func (t transcript) toolNames() []string {
	var names []string
	for _, m := range t.Messages {
		for _, tc := range m.Content.ToolCalls {
			if name, ok := tc["name"].(string); ok {
				names = append(names, name)
			}
		}
	}
	return names
}

// distinctGenerations counts unique plan generation values across all plan
// steps. Used by the `plan_generations` assertion (P2 iteration). Returns 0
// when there are no plan steps (no propose_plan was called).
func (t transcript) distinctGenerations() int {
	seen := map[int]bool{}
	for _, s := range t.PlanSteps {
		seen[s.Generation] = true
	}
	return len(seen)
}

type sessionState struct {
	ID         string `json:"id"`
	Status     string `json:"status"`
	Outcome    string `json:"outcome"`
	LastActive string `json:"last_active_at"`
}

// fetchTranscript fetches the messages from /sessions/{id} (which returns
// only session_id + messages) and the session metadata from /sessions
// (which returns status/outcome/last_active_at for each session). P5 also
// fetches /sessions/{id}/plan for the plan_generations assertion.
func fetchTranscript(ctx context.Context, gateway, sid string) (transcript, sessionState, error) {
	var t transcript
	resp, err := http.Get(gateway + "/sessions/" + sid)
	if err != nil {
		return t, sessionState{}, err
	}
	defer resp.Body.Close()
	b, err := io.ReadAll(resp.Body)
	if err != nil {
		return t, sessionState{}, err
	}
	if err := json.Unmarshal(b, &t); err != nil {
		return t, sessionState{}, err
	}
	// Fetch the plan (steps with generation numbers) for the
	// plan_generations assertion. A 404 or empty response is fine — a
	// pure-DB Q&A with no propose_plan has no plan.
	if planResp, perr := http.Get(gateway + "/sessions/" + sid + "/plan"); perr == nil {
		if planResp.StatusCode == 200 {
			pb, _ := io.ReadAll(planResp.Body)
			_ = json.Unmarshal(pb, &t) // fills t.PlanSteps via "steps" field
		}
		planResp.Body.Close()
	}
	// The detail endpoint doesn't return status/outcome — fetch from the
	// sessions list and find the matching id.
	s, err := fetchSessionMeta(ctx, gateway, sid)
	return t, s, err
}

// fetchSessionMeta fetches /sessions and extracts the one matching sid.
func fetchSessionMeta(ctx context.Context, gateway, sid string) (sessionState, error) {
	resp, err := http.Get(gateway + "/sessions")
	if err != nil {
		return sessionState{}, err
	}
	defer resp.Body.Close()
	var list struct {
		Sessions []sessionState `json:"sessions"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&list); err != nil {
		return sessionState{}, err
	}
	for _, s := range list.Sessions {
		if s.ID == sid {
			return s, nil
		}
	}
	return sessionState{}, fmt.Errorf("session %s not found in list", sid)
}

// convResult is the outcome of one conversation.
type convResult struct {
	SessionID     string
	Passed        bool
	Duration      time.Duration
	ToolCallCount int
	Assertions    []assertionResult
}

type assertionResult struct {
	Name   string
	Passed bool
	Detail string
}