feat: add /deploy-plugins webhook route + deploy-plugins script
This commit is contained in:
1002
cmd/nomos/agent.go
1002
cmd/nomos/agent.go
File diff suppressed because it is too large
Load Diff
@@ -1,402 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/dtoro/oikos/internal/nomos/session"
|
||||
"github.com/dtoro/oikos/internal/safego"
|
||||
"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
|
||||
}
|
||||
|
||||
// idleTaskThreshold is how long a goal-bearing session can sit non-terminal
|
||||
// with no activity before the idle sweep nudges it, per
|
||||
// plans/2026-07-11-task-completion-safety-net.md. Arbitrary starting point,
|
||||
// not measured against real task durations — long enough that it won't fire
|
||||
// mid-turn, short enough the board doesn't lie for hours.
|
||||
const idleTaskThreshold = 15 * time.Minute
|
||||
|
||||
// runIdleSweepWorker is the safety net for case 2 of
|
||||
// plans/2026-07-11-task-completion-safety-net.md: sessions that called
|
||||
// set_goal (so the inline safety net in agent.go correctly left them alone,
|
||||
// since they framed themselves as a real task) but then stalled without
|
||||
// ever calling complete_task. Coarser than runContinuationWorker's 4s tick
|
||||
// since "gone idle" is a much slower signal than "an execution just
|
||||
// finished." Blocks until ctx is cancelled.
|
||||
func (a *agent) runIdleSweepWorker(ctx context.Context) {
|
||||
if a.store == nil {
|
||||
slog.Warn("nomos: idle sweep worker disabled (no store)")
|
||||
return
|
||||
}
|
||||
slog.Info("nomos: idle sweep worker started")
|
||||
ticker := time.NewTicker(2 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
a.processIdleSweep(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// processIdleSweep nudges a stalled goal-bearing session once; if it's still
|
||||
// non-terminal on the NEXT sweep (meaning the nudge itself went unanswered,
|
||||
// not just that the model is still working), auto-closes it with a
|
||||
// visible "auto-closed" outcome instead of leaving it stuck forever — same
|
||||
// reasoning resumeSession already applies below for a different failure
|
||||
// mode (a resume that produces no response at all).
|
||||
func (a *agent) processIdleSweep(ctx context.Context) {
|
||||
stale := a.store.StaleGoalSessions(ctx, idleTaskThreshold, 5)
|
||||
for _, s := range stale {
|
||||
s := s
|
||||
if s.CompletionNudges == 0 {
|
||||
safego.Go("nomos:idle-nudge:"+s.ID, func() {
|
||||
note := fmt.Sprintf("[System: this task ('%s') has been idle for %s with no complete_task call. "+
|
||||
"If the goal is done (or can't be completed), call complete_task now with the outcome and a "+
|
||||
"one-line summary. If you're still genuinely working through the plan, ignore this and continue.]",
|
||||
s.Goal, idleTaskThreshold)
|
||||
note = a.store.EnrichResumeNote(ctx, s.ID, note)
|
||||
// P1: only count the nudge if it actually delivered. resumeSession
|
||||
// skips (returns false) when a turn is already active; bumping the
|
||||
// counter anyway would make the next sweep auto-close a merely-busy
|
||||
// session as "unanswered."
|
||||
if a.resumeSession(ctx, s.ID, note) {
|
||||
if err := a.store.BumpCompletionNudge(ctx, s.ID); err != nil {
|
||||
slog.Error("nomos: idle nudge bump failed", "session", s.ID, "error", err)
|
||||
}
|
||||
}
|
||||
})
|
||||
continue
|
||||
}
|
||||
safego.Go("nomos:idle-autoclose:"+s.ID, func() {
|
||||
summary := fmt.Sprintf("Auto-closed after %s idle with no response to a completion nudge.", idleTaskThreshold)
|
||||
if err := a.store.CompleteTask(ctx, s.ID, "partial", summary); err != nil {
|
||||
slog.Error("nomos: idle auto-close failed", "session", s.ID, "error", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// processContinuations dispatches each pending item as its OWN goroutine
|
||||
// (safego.Go, so a panic deep in one task's resumed turn — JSON parsing of
|
||||
// model output, an unexpected nil in a tool result — is recovered and logged
|
||||
// instead of taking down this whole function, which used to run every
|
||||
// item sequentially in the SAME goroutine as the ticker loop. Two problems
|
||||
// that fixed: (1) throughput — task B's continuation no longer waits for
|
||||
// task A's full (up to 10-minute) resumed turn to finish first, the exact
|
||||
// per-task blocking this session's earlier concurrency work removed from the
|
||||
// live-chat path but had left in place here; (2) survivability — since Go
|
||||
// panics unwind the goroutine they occur in, an unrecovered one here used to
|
||||
// mean this call (and every future tick, since the whole ticker loop runs in
|
||||
// one goroutine) would simply stop — auto-continuation for every task would
|
||||
// silently die until nomos restarted. Now a single bad item can only ever
|
||||
// take down its own goroutine.
|
||||
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) {
|
||||
// Re-open the assent window if this session is genuinely
|
||||
// executing (plan was approved, work is in progress) — the
|
||||
// window may have expired while the execution ran. Don't
|
||||
// penalize timing: the plan was approved, the work happened,
|
||||
// the result should flow back.
|
||||
sesh, seshErr := a.store.GetSession(ctx, p.SessionID)
|
||||
if seshErr == nil && sesh.Goal != "" && (sesh.Status == "executing" || sesh.Status == "planning") {
|
||||
a.openAssentWindow(ctx, p.SessionID)
|
||||
slog.Info("nomos: re-opened assent window for continuing session", "session", p.SessionID, "execution", p.ExecID)
|
||||
} else {
|
||||
// Genuinely no plan — inject a visible note so the
|
||||
// operator knows WHY the agent didn't auto-continue.
|
||||
note := fmt.Sprintf("[System: execution %s finished with status=%s, but the assent window for this session is not active. The agent will not auto-continue. Reply 'continue' or re-approve the plan to resume.]", p.ExecID, p.Status)
|
||||
body, _ := json.Marshal(map[string]any{"role": "assistant", "text": note, "auto": true})
|
||||
a.store.SaveMessage(context.Background(), p.SessionID, "assistant", body)
|
||||
a.store.MarkContinued(ctx, p.ExecID)
|
||||
continue
|
||||
}
|
||||
}
|
||||
// markContinued now happens inside continueSession, AFTER resumeSession
|
||||
// actually runs (P0). Pre-marking here consumed the item even when
|
||||
// resumeSession skipped on a busy session, losing the result.
|
||||
safego.Go("nomos:continue-session:"+p.SessionID, func() { 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 session.PendingContinuation) {
|
||||
slog.Info("nomos: auto-continuing session", "session", p.SessionID, "execution", p.ExecID, "status", p.Status)
|
||||
// P0 (plans/2026-08-03-nomos-chat-changes-review.md): mark the execution
|
||||
// continued ONLY after the turn actually ran. resumeSession skips (returns
|
||||
// false) when another turn is already active for this session; marking
|
||||
// before that — as the old code did — consumed the item (continued_at set,
|
||||
// never re-queued by pendingContinuations) and silently lost the result.
|
||||
// On a skip, leave it pending so the next worker tick retries once the
|
||||
// active turn frees the permit.
|
||||
if !a.resumeSession(ctx, p.SessionID, buildContinuationNote(p)) {
|
||||
slog.Info("nomos: continuation deferred — a turn is active; will retry next tick", "session", p.SessionID, "execution", p.ExecID)
|
||||
return
|
||||
}
|
||||
a.store.MarkContinued(ctx, p.ExecID)
|
||||
}
|
||||
|
||||
// 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.
|
||||
//
|
||||
// F1 (plan 2026-08-03): this is the single entry point for EVERY background
|
||||
// turn — the continuation worker, idle sweep, answer-question, /resume, and the
|
||||
// empty-message reconnect all funnel through here. It acquires the session's
|
||||
// turn permit non-blocking and SKIPS if a turn is already running. A duplicate
|
||||
// resume while a turn (live or background) is active is exactly the
|
||||
// interleaving that corrupted the activity panel and made tasks feel stuck.
|
||||
//
|
||||
// Returns whether the turn actually ran. Callers that mutate state before
|
||||
// resuming (the continuation worker's markContinued, the idle sweep's nudge
|
||||
// bump) MUST gate that mutation on a true return — otherwise a busy-skip leaves
|
||||
// the state changed but the work undone (lost continuation / false auto-close).
|
||||
// See plans/2026-08-03-nomos-chat-changes-review.md P0/P1.
|
||||
func (a *agent) resumeSession(ctx context.Context, sessionID, note string) bool {
|
||||
if !a.gate.Acquire(sessionID, 0) {
|
||||
slog.Info("nomos: turn already active, skipping background resume", "session", sessionID)
|
||||
return false
|
||||
}
|
||||
// Release the gate, then drain any operator message that was queued while
|
||||
// this background turn ran (plan 2026-08-03 F2). Queued messages are run as
|
||||
// real user turns server-side; resumeSession itself never enqueues.
|
||||
defer func() {
|
||||
a.gate.Release(sessionID)
|
||||
safego.Go("nomos:drain:"+sessionID, func() { a.drainQueued(context.Background(), sessionID) })
|
||||
}()
|
||||
|
||||
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
|
||||
var finalThinking 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,
|
||||
"thinking": finalThinking,
|
||||
"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()
|
||||
// B.3: escalate the recovery note across attempts — a transient flake
|
||||
// needs a different prompt than a model that's stuck no-op'ing. The
|
||||
// final attempt is maximally directive ("do this specific thing now").
|
||||
// B.5: back off between retries (4s, 8s) so a transient provider issue
|
||||
// has time to clear — 3 identical calls in 3 seconds just get 3
|
||||
// identical empties.
|
||||
notes := []string{
|
||||
note, // attempt 0: the original (already enriched per B.2) note
|
||||
fmt.Sprintf("[System: your previous turn produced no response. %s. Produce a response now — call the next tool or report progress in one sentence.]", note),
|
||||
fmt.Sprintf("[System: two consecutive empty responses. Stop trying to be clever. The next action is: pick the lowest-pending plan step, mark it running with update_plan_step, and call run for its target. Do that now.]"),
|
||||
}
|
||||
for attempt := 0; attempt < 3; attempt++ {
|
||||
if attempt > 0 {
|
||||
select {
|
||||
case <-cctx.Done():
|
||||
return true // a turn ran on an earlier attempt; consume, don't re-loop
|
||||
case <-time.After(time.Duration(2<<attempt) * time.Second): // 4s, 8s
|
||||
}
|
||||
}
|
||||
toolCalls, finalText, errText = nil, "", ""
|
||||
finalThinking = ""
|
||||
// P3: accumulate per-iteration reasoning instead of overwriting
|
||||
// (same fix as main.go's chat handler). Without this, a resumed
|
||||
// turn's intermediate thinking is lost on reload.
|
||||
// (same fix as main.go's chat handler). Without this, a resumed
|
||||
// turn's intermediate thinking is lost on reload.
|
||||
var textParts []string
|
||||
var thinkingParts []string
|
||||
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
|
||||
// One entry per tool call: tool_use creates it,
|
||||
// tool_result merges the result into the same entry
|
||||
// (matched by id). Before this fix, both events
|
||||
// appended separate entries, doubling every tool call.
|
||||
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: a poller sees this step land within seconds
|
||||
}
|
||||
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()
|
||||
}
|
||||
}
|
||||
if ev.Type == "error" {
|
||||
errText, _ = ev.Data.(string)
|
||||
}
|
||||
}
|
||||
a.chatWith(cctx, sessionID, "", notes[attempt], emit)
|
||||
if finalText != "" || len(toolCalls) > 0 {
|
||||
break
|
||||
}
|
||||
if attempt < 2 {
|
||||
slog.Warn("nomos: resume produced nothing, retrying", "session", sessionID, "error", errText, "attempt", attempt+1)
|
||||
}
|
||||
}
|
||||
|
||||
if errText != "" && finalText == "" {
|
||||
slog.Error("nomos: resume produced no response after retry", "session", sessionID, "error", errText)
|
||||
// Persist a visible system note in the transcript so the
|
||||
// operator sees what happened, but do NOT auto-complete the
|
||||
// task — leave it in 'executing' so a follow-up chat message
|
||||
// can resume it. Before this fix, the task was marked 'failed'
|
||||
// here, which ended it permanently and required starting over.
|
||||
resumeFailedNote := fmt.Sprintf("[System: auto-resume failed after retrying: %s. The task is paused — send another message to continue.]", errText)
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"role": "assistant",
|
||||
"text": resumeFailedNote,
|
||||
"auto": true,
|
||||
})
|
||||
if msgID != uuid.Nil {
|
||||
a.store.UpdateMessage(context.Background(), msgID, body)
|
||||
} else {
|
||||
// No placeholder was inserted (rare), save directly.
|
||||
a.store.SaveMessage(context.Background(), sessionID, "assistant", body)
|
||||
}
|
||||
return true // do not call persist() again — already persisted above
|
||||
}
|
||||
persist() // final state — same row, updated one last time with the concluding text
|
||||
return true
|
||||
}
|
||||
|
||||
// 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 session.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()
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/dtoro/oikos/internal/nomos/session"
|
||||
"github.com/dtoro/oikos/internal/nomos/turngate"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
func TestExtractExecutionIDs(t *testing.T) {
|
||||
// Real tool-result phrasings that should yield an execution id.
|
||||
pos := map[string]string{
|
||||
`"pct_create on host:strong auto-approved via assent window — execution 019f4b19-eafd-74ed-baa6-d24a27b3f52c running."`: "019f4b19-eafd-74ed-baa6-d24a27b3f52c",
|
||||
`"run on lxc:caddy requires approval (risk: config_mutation) — execution 019f4af7-7eff-7723-b38c-b540b267f407 queued."`: "019f4af7-7eff-7723-b38c-b540b267f407",
|
||||
`"apt_upgrade on host:hubris auto-approved via assent window — execution 019f4b58-c88c-7767-87dd-044608ced913 running."`: "019f4b58-c88c-7767-87dd-044608ced913",
|
||||
}
|
||||
for in, want := range pos {
|
||||
ids := extractExecutionIDs(in)
|
||||
if len(ids) != 1 || ids[0].String() != want {
|
||||
t.Errorf("extractExecutionIDs(%q) = %v, want [%s]", in, ids, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Synchronous auto-run and read-only results carry no "execution <uuid>"
|
||||
// phrasing — they've already completed inline and must NOT be linked for
|
||||
// continuation.
|
||||
neg := []string{
|
||||
`"run on host:strong (read_only, auto): 09:30 up 8 days"`,
|
||||
`"run on lxc:caddy (config_mutation, auto via assent window): done"`,
|
||||
`[{"slug":"lxc:caddy","health":"healthy"}]`,
|
||||
`"target not found: lxc:nope"`,
|
||||
}
|
||||
for _, in := range neg {
|
||||
if ids := extractExecutionIDs(in); len(ids) != 0 {
|
||||
t.Errorf("extractExecutionIDs(%q) = %v, want none", in, ids)
|
||||
}
|
||||
}
|
||||
|
||||
// De-dupes repeated ids in one result.
|
||||
dup := `execution 019f4b19-eafd-74ed-baa6-d24a27b3f52c queued ... execution 019f4b19-eafd-74ed-baa6-d24a27b3f52c running`
|
||||
if ids := extractExecutionIDs(dup); len(ids) != 1 {
|
||||
t.Errorf("expected de-dup to 1 id, got %v", ids)
|
||||
}
|
||||
}
|
||||
|
||||
// TestResumeSession_SkipsWhenBusy guards the P0 fix
|
||||
// (plans/2026-08-03-nomos-chat-changes-review.md): resumeSession must skip —
|
||||
// return false, body never executed — when a turn is already active for the
|
||||
// session. continueSession relies on this so it only marks a continuation
|
||||
// "continued" after a turn really ran (otherwise the result is lost: marked
|
||||
// continued, never re-queued by PendingContinuations).
|
||||
//
|
||||
// A minimal agent with only a gate is enough: if the body ever ran, chatWith
|
||||
// would dereference the nil provider and panic. Returning false cleanly proves
|
||||
// the body was skipped.
|
||||
func TestResumeSession_SkipsWhenBusy(t *testing.T) {
|
||||
a := &agent{gate: turngate.New()}
|
||||
if !a.gate.Acquire("sess", 0) {
|
||||
t.Fatal("precondition: initial acquire should succeed on a free session")
|
||||
}
|
||||
ran := a.resumeSession(context.Background(), "sess", "note")
|
||||
if ran {
|
||||
t.Fatal("resumeSession must return false (skip) while a turn is active for the session")
|
||||
}
|
||||
}
|
||||
|
||||
// TestContinueSession_DefersWhenBusy guards the other half of P0: when the
|
||||
// session is busy, continueSession defers (leaves the execution pending for the
|
||||
// next worker tick) instead of running or marking it. It must return cleanly
|
||||
// without reaching resumeSession's body (nil provider → panic) or markContinued.
|
||||
func TestContinueSession_DefersWhenBusy(t *testing.T) {
|
||||
a := &agent{gate: turngate.New()}
|
||||
if !a.gate.Acquire("sess", 0) {
|
||||
t.Fatal("precondition: initial acquire should succeed on a free session")
|
||||
}
|
||||
p := session.PendingContinuation{ExecID: uuid.New(), SessionID: "sess", Status: "completed"}
|
||||
a.continueSession(context.Background(), p) // must not panic; must not run/mark
|
||||
}
|
||||
@@ -1,372 +0,0 @@
|
||||
// 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. ?all=true returns every
|
||||
// generation so the assertion can count them (the default view returns
|
||||
// only the current generation).
|
||||
if planResp, perr := http.Get(gateway + "/sessions/" + sid + "/plan?all=true"); 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
|
||||
}
|
||||
@@ -1,236 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
// conversation is one golden conversation from a manifest.
|
||||
type conversation struct {
|
||||
Name string `yaml:"name"`
|
||||
Prompt string `yaml:"prompt"`
|
||||
Followup string `yaml:"followup"` // backward compat: single followup
|
||||
Followups []string `yaml:"followups"` // P5: multi-turn followups
|
||||
Assertions []assertion `yaml:"assertions"`
|
||||
}
|
||||
|
||||
// followups returns the full list of follow-up messages, supporting both
|
||||
// the single `followup` field (backward compat) and the multi-turn
|
||||
// `followups` list.
|
||||
func (c conversation) followups() []string {
|
||||
if len(c.Followups) > 0 {
|
||||
return c.Followups
|
||||
}
|
||||
if c.Followup != "" {
|
||||
return []string{c.Followup}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// assertion is one check against the final transcript. The `kind` field
|
||||
// selects the scorer; the rest are scorer-specific parameters.
|
||||
//
|
||||
// Supported kinds:
|
||||
//
|
||||
// completes — session status reached done/failed (not stuck executing)
|
||||
// outcome_is — session outcome == value (success/failure/partial)
|
||||
// no_propose_plan — propose_plan was never called
|
||||
// proposes_plan — propose_plan called >= 1 time (plan-always model; P1)
|
||||
// proposes_plan_once — propose_plan was called exactly once
|
||||
// no_duplicate_proposal — propose_plan called at most once
|
||||
// plan_before_run — the first `run` call comes after the first `propose_plan` (P1 ordering gate)
|
||||
// plan_generations — the persisted plan has exactly `value` distinct generations (P2 iteration: 1 = single, 2 = one followup)
|
||||
// writes_back — update_entity_attributes or create_relationship was called
|
||||
// max_tool_calls — total tool calls <= value
|
||||
// max_run_calls — total `run` calls <= value
|
||||
// no_run — `run` was never called
|
||||
// calls_tool — the named tool appears in the transcript
|
||||
// plan_step_count — the plan has exactly `value` steps
|
||||
// no_duplicate_complete — complete_task called at most once
|
||||
type assertion struct {
|
||||
Kind string `yaml:"kind"`
|
||||
Value any `yaml:"value"`
|
||||
}
|
||||
|
||||
// loadManifest reads a YAML file containing a list of conversations.
|
||||
func loadManifest(path string) ([]conversation, error) {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var convs []conversation
|
||||
if err := yaml.Unmarshal(b, &convs); err != nil {
|
||||
return nil, fmt.Errorf("parse %s: %w", path, err)
|
||||
}
|
||||
return convs, nil
|
||||
}
|
||||
|
||||
// scoreAssertions evaluates each assertion against the transcript + session.
|
||||
func scoreAssertions(asserts []assertion, t transcript, s sessionState) []assertionResult {
|
||||
out := make([]assertionResult, 0, len(asserts))
|
||||
for _, a := range asserts {
|
||||
r := assertionResult{Name: a.Kind}
|
||||
r.Passed, r.Detail = scoreOne(a, t, s)
|
||||
if !r.Passed && r.Detail == "" {
|
||||
r.Detail = "assertion failed"
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func scoreOne(a assertion, t transcript, s sessionState) (bool, string) {
|
||||
tools := t.toolNames()
|
||||
switch a.Kind {
|
||||
case "completes":
|
||||
if s.Status == "done" || s.Status == "failed" {
|
||||
return true, fmt.Sprintf("status=%s", s.Status)
|
||||
}
|
||||
return false, fmt.Sprintf("status=%s (not terminal)", s.Status)
|
||||
|
||||
case "outcome_is":
|
||||
want, _ := a.Value.(string)
|
||||
if s.Outcome == want {
|
||||
return true, fmt.Sprintf("outcome=%s", s.Outcome)
|
||||
}
|
||||
return false, fmt.Sprintf("outcome=%s, want %s", s.Outcome, want)
|
||||
|
||||
case "no_propose_plan":
|
||||
n := countTool(tools, "propose_plan")
|
||||
if n == 0 {
|
||||
return true, "propose_plan not called"
|
||||
}
|
||||
return false, fmt.Sprintf("propose_plan called %d time(s)", n)
|
||||
|
||||
case "proposes_plan":
|
||||
// P1 plan-always: propose_plan called >= 1 time.
|
||||
n := countTool(tools, "propose_plan")
|
||||
if n >= 1 {
|
||||
return true, fmt.Sprintf("propose_plan called %d time(s)", n)
|
||||
}
|
||||
return false, "propose_plan never called (plan-always requires >= 1)"
|
||||
|
||||
case "proposes_plan_once":
|
||||
n := countTool(tools, "propose_plan")
|
||||
if n == 1 {
|
||||
return true, "propose_plan called once"
|
||||
}
|
||||
return false, fmt.Sprintf("propose_plan called %d time(s), want 1", n)
|
||||
|
||||
case "no_duplicate_proposal":
|
||||
n := countTool(tools, "propose_plan")
|
||||
if n <= 1 {
|
||||
return true, fmt.Sprintf("propose_plan called %d time(s)", n)
|
||||
}
|
||||
return false, fmt.Sprintf("propose_plan called %d time(s), want <= 1", n)
|
||||
|
||||
case "plan_before_run":
|
||||
// P1 ordering gate: the first `run` call's global index in the
|
||||
// transcript is strictly greater than the first `propose_plan`
|
||||
// index. Both indices are over the flat tool-call list (across all
|
||||
// messages, in order).
|
||||
planIdx, runIdx := -1, -1
|
||||
for i, name := range tools {
|
||||
if name == "propose_plan" && planIdx == -1 {
|
||||
planIdx = i
|
||||
}
|
||||
if name == "run" && runIdx == -1 {
|
||||
runIdx = i
|
||||
}
|
||||
}
|
||||
if runIdx == -1 {
|
||||
return true, "run never called (ordering trivially satisfied)"
|
||||
}
|
||||
if planIdx == -1 {
|
||||
return false, "run called but propose_plan never called"
|
||||
}
|
||||
if planIdx < runIdx {
|
||||
return true, fmt.Sprintf("propose_plan at index %d before run at index %d", planIdx, runIdx)
|
||||
}
|
||||
return false, fmt.Sprintf("run at index %d before propose_plan at index %d", runIdx, planIdx)
|
||||
|
||||
case "plan_generations":
|
||||
// P2 iteration: counts distinct `generation` values in
|
||||
// session_plan_steps. 1 = single sub-task, 2 = one follow-up
|
||||
// sub-task, etc. Requires the plan endpoint to return generation
|
||||
// values; the eval fetches /sessions/{id}/plan and passes it via
|
||||
// the transcript's PlanSteps field.
|
||||
want := toInt(a.Value)
|
||||
gens := t.distinctGenerations()
|
||||
if gens == want {
|
||||
return true, fmt.Sprintf("%d plan generation(s)", gens)
|
||||
}
|
||||
return false, fmt.Sprintf("%d plan generation(s), want %d", gens, want)
|
||||
|
||||
case "writes_back":
|
||||
n := countTool(tools, "update_entity_attributes") + countTool(tools, "create_relationship")
|
||||
if n > 0 {
|
||||
return true, fmt.Sprintf("%d writeback call(s)", n)
|
||||
}
|
||||
return false, "no update_entity_attributes or create_relationship calls"
|
||||
|
||||
case "max_tool_calls":
|
||||
max := toInt(a.Value)
|
||||
if t.toolCallCount() <= max {
|
||||
return true, fmt.Sprintf("%d tool calls (<= %d)", t.toolCallCount(), max)
|
||||
}
|
||||
return false, fmt.Sprintf("%d tool calls, want <= %d", t.toolCallCount(), max)
|
||||
|
||||
case "max_run_calls":
|
||||
max := toInt(a.Value)
|
||||
n := countTool(tools, "run")
|
||||
if n <= max {
|
||||
return true, fmt.Sprintf("%d run calls (<= %d)", n, max)
|
||||
}
|
||||
return false, fmt.Sprintf("%d run calls, want <= %d", n, max)
|
||||
|
||||
case "no_run":
|
||||
n := countTool(tools, "run")
|
||||
if n == 0 {
|
||||
return true, "run not called"
|
||||
}
|
||||
return false, fmt.Sprintf("run called %d time(s)", n)
|
||||
|
||||
case "calls_tool":
|
||||
want, _ := a.Value.(string)
|
||||
n := countTool(tools, want)
|
||||
if n > 0 {
|
||||
return true, fmt.Sprintf("%s called %d time(s)", want, n)
|
||||
}
|
||||
return false, fmt.Sprintf("%s not called", want)
|
||||
|
||||
case "no_duplicate_complete":
|
||||
n := countTool(tools, "complete_task")
|
||||
if n <= 1 {
|
||||
return true, fmt.Sprintf("complete_task called %d time(s)", n)
|
||||
}
|
||||
return false, fmt.Sprintf("complete_task called %d time(s), want <= 1", n)
|
||||
|
||||
default:
|
||||
return false, fmt.Sprintf("unknown assertion kind: %s", a.Kind)
|
||||
}
|
||||
}
|
||||
|
||||
func countTool(names []string, name string) int {
|
||||
n := 0
|
||||
for _, x := range names {
|
||||
if x == name {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func toInt(v any) int {
|
||||
switch x := v.(type) {
|
||||
case int:
|
||||
return x
|
||||
case int64:
|
||||
return int(x)
|
||||
case float64:
|
||||
return int(x)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// sseReader parses a text/event-stream into a sequence of JSON events.
|
||||
// Each event is one or more "data: " lines; the lines are concatenated
|
||||
// and parsed as a single JSON object. Blank lines separate events.
|
||||
type sseReader struct {
|
||||
r *bufio.Reader
|
||||
}
|
||||
|
||||
func newSSEReader(r io.Reader) *sseReader {
|
||||
return &sseReader{r: bufio.NewReader(r)}
|
||||
}
|
||||
|
||||
func (s *sseReader) next() (map[string]any, error) {
|
||||
var data strings.Builder
|
||||
for {
|
||||
line, err := s.r.ReadString('\n')
|
||||
if err != nil {
|
||||
if err == io.EOF && data.Len() > 0 {
|
||||
return parseEvent(data.String())
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
line = strings.TrimRight(line, "\r\n")
|
||||
if line == "" {
|
||||
if data.Len() > 0 {
|
||||
return parseEvent(data.String())
|
||||
}
|
||||
continue // blank line, no event buffered yet
|
||||
}
|
||||
if strings.HasPrefix(line, "data: ") {
|
||||
data.WriteString(strings.TrimPrefix(line, "data: "))
|
||||
} else if strings.HasPrefix(line, "data:") {
|
||||
data.WriteString(strings.TrimPrefix(line, "data:"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func parseEvent(s string) (map[string]any, error) {
|
||||
var ev map[string]any
|
||||
if err := json.Unmarshal([]byte(s), &ev); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ev, nil
|
||||
}
|
||||
348
cmd/nomos/mcp.go
348
cmd/nomos/mcp.go
@@ -1,348 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ─── MCP Streamable HTTP client ────────────────────────────────────────
|
||||
|
||||
type mcpClient struct {
|
||||
baseURL string
|
||||
token string // OIKOS_MCP_BEARER_TOKEN — api's combinedAuth requires it on every request (no dev-open bypass)
|
||||
sessionID string
|
||||
http *http.Client
|
||||
nextID int
|
||||
mu sync.Mutex // one client serializes its own MCP calls (the pool gives each session its own client, so this never blocks another session)
|
||||
// toolsCache holds the last tools/list result. The tool list is static
|
||||
// for the lifetime of one MCP connection — it only changes when the api
|
||||
// process (re)registers tools, i.e. on a restart, which this client
|
||||
// already detects and reacts to via reconnectLocked. Without this,
|
||||
// buildTools (called at the start of EVERY chat turn, including every
|
||||
// auto-continuation resume) paid a full tools/list round-trip every
|
||||
// single time for a list that's almost always identical to the last one.
|
||||
// Guarded separately from mu (not reused) so a cache check never
|
||||
// contends with an in-flight doRequest call for a different method.
|
||||
toolsMu sync.Mutex
|
||||
toolsCache []toolDef
|
||||
}
|
||||
|
||||
func newMCPClient(baseURL, token string) (*mcpClient, error) {
|
||||
c := &mcpClient{
|
||||
baseURL: baseURL,
|
||||
token: token,
|
||||
http: &http.Client{Timeout: 120 * time.Second},
|
||||
}
|
||||
|
||||
resp, err := c.doRequest("initialize", map[string]any{
|
||||
"protocolVersion": "2024-11-05",
|
||||
"capabilities": map[string]any{},
|
||||
"clientInfo": map[string]any{"name": "nomos", "version": "2.0"},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("initialize: %w", err)
|
||||
}
|
||||
if resp.sessionID == "" {
|
||||
return nil, fmt.Errorf("no session ID in initialize response")
|
||||
}
|
||||
c.sessionID = resp.sessionID
|
||||
|
||||
c.doRequest("notifications/initialized", map[string]any{})
|
||||
|
||||
slog.Info("nomos: mcp connected", "session", c.sessionID[:16]+"...")
|
||||
return c, nil
|
||||
}
|
||||
|
||||
type mcpJSONRPCResponse struct {
|
||||
sessionID string
|
||||
Result json.RawMessage `json:"result"`
|
||||
Error json.RawMessage `json:"error"`
|
||||
}
|
||||
|
||||
// errStaleSession signals that the MCP server rejected our session id (e.g.
|
||||
// after an api/MCP restart), so the client should re-initialize and retry.
|
||||
var errStaleSession = fmt.Errorf("mcp session stale")
|
||||
|
||||
// doRequest serializes MCP calls and transparently re-initializes the session
|
||||
// if the server has forgotten it (common after an api redeploy), retrying the
|
||||
// original call once. Without this, an api restart permanently breaks nomos
|
||||
// until it is itself restarted.
|
||||
func (c *mcpClient) doRequest(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
resp, err := c.send(method, params)
|
||||
if err != nil && method != "initialize" && errors.Is(err, errStaleSession) {
|
||||
slog.Warn("nomos: mcp session stale, reconnecting")
|
||||
if rerr := c.reconnectLocked(); rerr != nil {
|
||||
return nil, fmt.Errorf("mcp reconnect: %w (original: %v)", rerr, err)
|
||||
}
|
||||
return c.send(method, params)
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
|
||||
// reconnectLocked re-initializes the MCP session. The caller must hold c.mu.
|
||||
func (c *mcpClient) reconnectLocked() error {
|
||||
c.sessionID = ""
|
||||
// A reconnect means the api process was restarted (or forgot us) — its
|
||||
// tool registration may have changed, so the cached list is no longer
|
||||
// trustworthy.
|
||||
c.toolsMu.Lock()
|
||||
c.toolsCache = nil
|
||||
c.toolsMu.Unlock()
|
||||
resp, err := c.send("initialize", map[string]any{
|
||||
"protocolVersion": "2024-11-05",
|
||||
"capabilities": map[string]any{},
|
||||
"clientInfo": map[string]any{"name": "nomos", "version": "2.0"},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if resp.sessionID == "" {
|
||||
return fmt.Errorf("no session ID on re-initialize")
|
||||
}
|
||||
c.sessionID = resp.sessionID
|
||||
_, _ = c.send("notifications/initialized", map[string]any{})
|
||||
slog.Info("nomos: mcp reconnected", "session", c.sessionID[:16]+"...")
|
||||
return nil
|
||||
}
|
||||
|
||||
// send performs one MCP round-trip. It does not lock; callers hold c.mu.
|
||||
func (c *mcpClient) send(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
|
||||
c.nextID++
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"method": method,
|
||||
"params": params,
|
||||
"id": c.nextID,
|
||||
})
|
||||
|
||||
req, err := http.NewRequest(http.MethodPost, c.baseURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json, text/event-stream")
|
||||
if c.sessionID != "" {
|
||||
req.Header.Set("Mcp-Session-Id", c.sessionID)
|
||||
}
|
||||
if c.token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+c.token)
|
||||
}
|
||||
|
||||
resp, err := c.http.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// A rejected/unknown session comes back as 4xx (commonly 400/404).
|
||||
if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusBadRequest {
|
||||
return nil, errStaleSession
|
||||
}
|
||||
|
||||
result := &mcpJSONRPCResponse{}
|
||||
result.sessionID = resp.Header.Get("Mcp-Session-Id")
|
||||
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
|
||||
gotData := false
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if strings.HasPrefix(line, "data: ") {
|
||||
gotData = true
|
||||
data := line[6:]
|
||||
if err := json.Unmarshal([]byte(data), result); err != nil {
|
||||
return nil, fmt.Errorf("parse response: %w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if result.Error != nil {
|
||||
return nil, fmt.Errorf("rpc error: %s", string(result.Error))
|
||||
}
|
||||
|
||||
// Empty body with no result and a session set: the server likely dropped
|
||||
// our session. Notifications legitimately return no data, so exempt them.
|
||||
if !gotData && result.Result == nil && method != "notifications/initialized" {
|
||||
return nil, errStaleSession
|
||||
}
|
||||
|
||||
if result.sessionID != "" {
|
||||
c.sessionID = result.sessionID
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (c *mcpClient) callTool(name string, args map[string]any) (any, error) {
|
||||
resp, err := c.doRequest("tools/call", map[string]any{
|
||||
"name": name,
|
||||
"arguments": args,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var toolResult struct {
|
||||
Content []struct {
|
||||
Type string `json:"type"`
|
||||
Text string `json:"text"`
|
||||
} `json:"content"`
|
||||
}
|
||||
if err := json.Unmarshal(resp.Result, &toolResult); err != nil {
|
||||
return string(resp.Result), nil
|
||||
}
|
||||
|
||||
var texts []string
|
||||
for _, c := range toolResult.Content {
|
||||
if c.Type == "text" {
|
||||
var parsed any
|
||||
if json.Unmarshal([]byte(c.Text), &parsed) == nil {
|
||||
return parsed, nil
|
||||
}
|
||||
texts = append(texts, c.Text)
|
||||
}
|
||||
}
|
||||
if len(texts) == 1 {
|
||||
return texts[0], nil
|
||||
}
|
||||
return texts, nil
|
||||
}
|
||||
|
||||
func (c *mcpClient) listTools() ([]string, error) {
|
||||
resp, err := c.doRequest("tools/list", map[string]any{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var tr struct {
|
||||
Tools []struct {
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
} `json:"tools"`
|
||||
}
|
||||
if err := json.Unmarshal(resp.Result, &tr); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var names []string
|
||||
for _, t := range tr.Tools {
|
||||
names = append(names, t.Name)
|
||||
}
|
||||
return names, nil
|
||||
}
|
||||
|
||||
func (c *mcpClient) close() {
|
||||
}
|
||||
|
||||
// ─── Per-session MCP client pool ────────────────────────────────────────
|
||||
//
|
||||
// A single shared mcpClient serializes EVERY tool call across EVERY
|
||||
// concurrently-running task through one mutex (see mcpClient.mu) — `run`
|
||||
// executes its SSH command synchronously inside that lock and is capped at
|
||||
// up to 10 minutes, so one task mid-`run` stalled every other task's tool
|
||||
// calls, even trivial reads, behind it. The MCP *server* has no per-
|
||||
// connection state to protect (newServer in internal/mcp/server.go returns
|
||||
// one shared *mcp.Server instance whose tool handlers close only over the DB
|
||||
// pool, which is already safe for concurrent use) — the mutex existed purely
|
||||
// because the *client* reused one stateful transport session, not because
|
||||
// the server needed it. Giving each task's own session its own client
|
||||
// removes the cross-task serialization entirely: a task's own tool calls
|
||||
// stay sequential (which they already are — the agent loop calls tools one
|
||||
// at a time within a turn), but no longer block anyone else's.
|
||||
type mcpClientPool struct {
|
||||
baseURL string
|
||||
token string
|
||||
mu sync.Mutex
|
||||
clients map[string]*pooledMCPClient
|
||||
}
|
||||
|
||||
type pooledMCPClient struct {
|
||||
client *mcpClient
|
||||
lastUsed time.Time
|
||||
}
|
||||
|
||||
func newMCPClientPool(baseURL, token string) *mcpClientPool {
|
||||
return &mcpClientPool{baseURL: baseURL, token: token, clients: make(map[string]*pooledMCPClient)}
|
||||
}
|
||||
|
||||
// get returns the client for sessionID, creating and initializing one (a
|
||||
// real MCP handshake) on first use. Session ids that don't identify a real
|
||||
// persisted conversation ("" / "ephemeral", the no-DB-store path; "query",
|
||||
// the structured /query endpoint) still get exactly one dedicated,
|
||||
// reused client each via the same map — just keyed on a fixed string instead
|
||||
// of a real session id — so that traffic doesn't pay a fresh handshake per
|
||||
// request while still never sharing a connection with an actual task.
|
||||
func (p *mcpClientPool) get(sessionID string) (*mcpClient, error) {
|
||||
key := sessionID
|
||||
if key == "" {
|
||||
key = "ephemeral"
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
if pc, ok := p.clients[key]; ok {
|
||||
pc.lastUsed = time.Now()
|
||||
p.mu.Unlock()
|
||||
return pc.client, nil
|
||||
}
|
||||
p.mu.Unlock()
|
||||
|
||||
// Initialize outside the lock — it's a network round-trip, and holding
|
||||
// the pool mutex for it would serialize unrelated sessions' first calls
|
||||
// behind each other, undermining the whole point of this pool.
|
||||
c, err := newMCPClient(p.baseURL, p.token)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
p.mu.Lock()
|
||||
// Another goroutine may have created one for the same key while we were
|
||||
// initializing (two of this session's tool calls racing on a cold
|
||||
// start); keep whichever won, close out the loser's connection (a no-op
|
||||
// today, but future-proof if mcpClient.close ever does real teardown).
|
||||
if existing, ok := p.clients[key]; ok {
|
||||
p.mu.Unlock()
|
||||
c.close()
|
||||
return existing.client, nil
|
||||
}
|
||||
p.clients[key] = &pooledMCPClient{client: c, lastUsed: time.Now()}
|
||||
p.mu.Unlock()
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// mcpClientIdleTimeout is how long an idle session's MCP client is kept
|
||||
// before eviction — long enough to outlive a single slow `run` (capped at 10
|
||||
// minutes server-side) plus normal think-time between a task's tool calls,
|
||||
// short enough not to accumulate one abandoned connection per finished task
|
||||
// forever.
|
||||
const mcpClientIdleTimeout = 20 * time.Minute
|
||||
|
||||
// sweep evicts clients idle past mcpClientIdleTimeout. Call on a ticker.
|
||||
func (p *mcpClientPool) sweep() {
|
||||
cutoff := time.Now().Add(-mcpClientIdleTimeout)
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for key, pc := range p.clients {
|
||||
if pc.lastUsed.Before(cutoff) {
|
||||
pc.client.close()
|
||||
delete(p.clients, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *mcpClientPool) closeAll() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for key, pc := range p.clients {
|
||||
pc.client.close()
|
||||
delete(p.clients, key)
|
||||
}
|
||||
}
|
||||
@@ -1,696 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/dtoro/oikos/internal/nomos/session"
|
||||
"github.com/dtoro/oikos/internal/safego"
|
||||
"github.com/dtoro/oikos/internal/secrets"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 {
|
||||
fmt.Fprintln(os.Stderr, "usage: nomos serve")
|
||||
os.Exit(1)
|
||||
}
|
||||
if os.Args[1] == "healthcheck" {
|
||||
runHealthcheck()
|
||||
return
|
||||
}
|
||||
mcpURL := os.Getenv("NOMOS_MCP_URL")
|
||||
if mcpURL == "" {
|
||||
mcpURL = "http://localhost:8090/mcp"
|
||||
}
|
||||
mcpToken := os.Getenv("OIKOS_MCP_BEARER_TOKEN")
|
||||
|
||||
agentSlug := os.Getenv("NOMOS_AGENT_SLUG")
|
||||
if agentSlug == "" {
|
||||
agentSlug = "agent:nomos"
|
||||
}
|
||||
|
||||
databaseURL := os.Getenv("DATABASE_URL")
|
||||
if databaseURL == "" {
|
||||
databaseURL = os.Getenv("OIKOS_DATABASE_URL")
|
||||
}
|
||||
|
||||
sec := secrets.NewManagerFromConfig(
|
||||
os.Getenv("OIKOS_INFISICAL_SITE_URL"),
|
||||
os.Getenv("OIKOS_INFISICAL_CLIENT_ID"),
|
||||
os.Getenv("OIKOS_INFISICAL_CLIENT_SECRET"),
|
||||
os.Getenv("OIKOS_INFISICAL_PROJECT_ID"),
|
||||
os.Getenv("OIKOS_INFISICAL_ENV"),
|
||||
os.Getenv("OIKOS_SECRETS_DIR"),
|
||||
)
|
||||
var openrouterAPIKey string
|
||||
var secretsResolved int
|
||||
if sec != nil {
|
||||
resCtx, resCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
if v := secrets.ResolveSecret(resCtx, sec, "mcp_bearer-token", ""); v != "" {
|
||||
mcpToken = v
|
||||
secretsResolved++
|
||||
}
|
||||
openrouterAPIKey = secrets.ResolveSecret(resCtx, sec, "openrouter_api-key", os.Getenv("OPENROUTER_API_KEY"))
|
||||
if openrouterAPIKey != "" && openrouterAPIKey != os.Getenv("OPENROUTER_API_KEY") {
|
||||
secretsResolved++
|
||||
}
|
||||
resCancel()
|
||||
if secretsResolved > 0 {
|
||||
slog.Info("nomos: secrets resolved from Infisical", "count", secretsResolved)
|
||||
}
|
||||
}
|
||||
|
||||
switch os.Args[1] {
|
||||
case "serve":
|
||||
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
defer cancel()
|
||||
|
||||
// One MCP client PER SESSION, not one shared client for the whole
|
||||
// process — see mcpClientPool's doc comment. A dedicated client is
|
||||
// created lazily on each session's first tool call.
|
||||
clientPool := newMCPClientPool(mcpURL, mcpToken)
|
||||
// Prove connectivity at startup the same way the old single-client
|
||||
// constructor did, so a misconfigured/unreachable MCP endpoint still
|
||||
// fails fast on boot instead of only on the first real chat. Doesn't
|
||||
// reuse the pool (nothing to key it by yet) — just a throwaway probe.
|
||||
if probe, err := newMCPClient(mcpURL, mcpToken); err != nil {
|
||||
slog.Error("nomos: mcp connect", "url", mcpURL, "error", err)
|
||||
os.Exit(1)
|
||||
} else {
|
||||
probe.close()
|
||||
}
|
||||
|
||||
st, err := session.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
slog.Error("nomos: db connect", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if st != nil {
|
||||
defer st.Close()
|
||||
}
|
||||
|
||||
nAgent, err := newAgent(ctx, clientPool, st, agentSlug, openrouterAPIKey)
|
||||
if err != nil {
|
||||
slog.Error("nomos: agent init", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Event-driven auto-continuation: feed finished async executions back
|
||||
// into the agent so an approved plan runs to completion (and recovers
|
||||
// from failures) without the operator ticking it forward each step.
|
||||
safego.Go("nomos:continuation-worker", func() { nAgent.runContinuationWorker(ctx) })
|
||||
|
||||
// Idle sweep for stalled goal-bearing tasks (fix 2+3 of
|
||||
// plans/2026-07-11-task-completion-safety-net.md) — a coarser,
|
||||
// slower-ticking counterpart to the continuation worker above.
|
||||
safego.Go("nomos:idle-sweep-worker", func() { nAgent.runIdleSweepWorker(ctx) })
|
||||
|
||||
safego.Go("nomos:mcp-pool-sweeper", func() {
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
clientPool.sweep()
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// Stale execution sweep: cancels non-terminal executions older than
|
||||
// 10 minutes (orphaned by MCP timeouts — see cleanupStaleExecutions).
|
||||
safego.Go("nomos:stale-execution-sweeper", func() {
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
st.CleanupStaleExecutions(ctx, 10*time.Minute)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(200)
|
||||
w.Write([]byte("ok"))
|
||||
})
|
||||
mux.HandleFunc("/query", func(w http.ResponseWriter, r *http.Request) {
|
||||
handleQuery(w, r, clientPool, agentSlug, mcpURL)
|
||||
})
|
||||
mux.HandleFunc("/chat", func(w http.ResponseWriter, r *http.Request) {
|
||||
handleChat(w, r, nAgent, st)
|
||||
})
|
||||
mux.HandleFunc("/sessions", func(w http.ResponseWriter, r *http.Request) {
|
||||
handleSessionsList(w, r, st)
|
||||
})
|
||||
mux.HandleFunc("/sessions/", func(w http.ResponseWriter, r *http.Request) {
|
||||
handleSessionDetail(w, r, st, nAgent)
|
||||
})
|
||||
|
||||
addr := os.Getenv("NOMOS_LISTEN")
|
||||
if addr == "" {
|
||||
addr = ":8092"
|
||||
}
|
||||
|
||||
srv := &http.Server{Addr: addr, Handler: mux}
|
||||
safego.Go("nomos:http-server", func() {
|
||||
slog.Info("nomos: gateway listening", "addr", addr, "mcp", mcpURL, "db", databaseURL != "")
|
||||
if err := srv.ListenAndServe(); err != http.ErrServerClosed {
|
||||
slog.Error("nomos: serve", "error", err)
|
||||
}
|
||||
})
|
||||
|
||||
<-ctx.Done()
|
||||
slog.Info("nomos: shutting down")
|
||||
srv.Shutdown(context.Background())
|
||||
clientPool.closeAll()
|
||||
|
||||
default:
|
||||
fmt.Fprintf(os.Stderr, "unknown command: %s\n", os.Args[1])
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func runHealthcheck() {
|
||||
addr := os.Getenv("NOMOS_LISTEN")
|
||||
if addr == "" {
|
||||
addr = ":8092"
|
||||
}
|
||||
host := addr
|
||||
if strings.HasPrefix(host, ":") {
|
||||
host = "127.0.0.1" + host
|
||||
}
|
||||
client := &http.Client{Timeout: 3 * time.Second}
|
||||
resp, err := client.Get("http://" + host + "/healthz")
|
||||
if err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func sseEvent(w http.ResponseWriter, flusher http.Flusher, event agentEvent) {
|
||||
data, _ := json.Marshal(event)
|
||||
fmt.Fprintf(w, "data: %s\n\n", data)
|
||||
flusher.Flush()
|
||||
}
|
||||
|
||||
func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *session.Store) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
|
||||
var req struct {
|
||||
SessionID string `json:"session_id"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "bad request: "+err.Error(), 400)
|
||||
return
|
||||
}
|
||||
if req.Message == "" && req.SessionID == "" {
|
||||
http.Error(w, "message is required", 400)
|
||||
return
|
||||
}
|
||||
|
||||
// Empty message with an existing session = reconnect/resume. This path is
|
||||
// defensive now — the frontend (post F2) recovers a dropped SSE via the
|
||||
// poller + terminal task.status clearing, and no longer POSTs empty
|
||||
// messages. If a client ever does, route into resumeSession so the agent
|
||||
// reports current state — but SKIP a terminal session (done/failed/
|
||||
// abandoned): there's nothing to resume, and running a "report state"
|
||||
// turn there is just a spare turn the operator never asked for (P2.1).
|
||||
if req.Message == "" && req.SessionID != "" {
|
||||
if sess, err := st.GetSession(context.Background(), req.SessionID); err == nil {
|
||||
switch sess.Status {
|
||||
case "done", "failed", "abandoned":
|
||||
slog.Info("nomos: reconnect skipped — session already terminal", "session", req.SessionID, "status", sess.Status)
|
||||
w.WriteHeader(202)
|
||||
return
|
||||
}
|
||||
}
|
||||
slog.Info("nomos: reconnect", "session", req.SessionID)
|
||||
safego.Go("nomos:reconnect:"+req.SessionID, func() {
|
||||
base := "[System: the operator's connection was re-established. The task may have progressed in the background.]"
|
||||
note := st.EnrichResumeNote(context.Background(), req.SessionID, base)
|
||||
a.resumeSession(context.Background(), req.SessionID, note)
|
||||
})
|
||||
// Return 202 so the frontend doesn't try to consume an SSE stream
|
||||
// from this POST — resumeSession writes to the DB directly and
|
||||
// the poller picks it up.
|
||||
w.WriteHeader(202)
|
||||
return
|
||||
}
|
||||
|
||||
flusher, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
http.Error(w, "streaming not supported", 500)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering
|
||||
w.WriteHeader(200)
|
||||
|
||||
// All writes to w (events + the keepalive comment below) go through one
|
||||
// mutex: http.ResponseWriter is NOT safe for concurrent use, and the
|
||||
// keepalive ticker runs alongside the turn's event sink (plan 2026-08-03
|
||||
// F3). Without this, interleaved writes corrupt the SSE stream.
|
||||
var writeMu sync.Mutex
|
||||
writeEvent := func(ev agentEvent) {
|
||||
writeMu.Lock()
|
||||
defer writeMu.Unlock()
|
||||
sseEvent(w, flusher, ev)
|
||||
}
|
||||
|
||||
ctx := r.Context()
|
||||
sessionID := req.SessionID
|
||||
|
||||
// pctx (persistence context) is deliberately context.Background(), not
|
||||
// ctx/r.Context(), for every DB write in this handler — ctx cancels the
|
||||
// instant the client disconnects (Stop button, tab close, network blip),
|
||||
// and a write made with an already-cancelled context fails. Before this
|
||||
// fix, the assistant message was only ever saved ONCE, at the very end,
|
||||
// using ctx — so a disconnect mid-turn silently lost the ENTIRE turn's
|
||||
// tool-call history from the persisted transcript, even though real work
|
||||
// (executions launched, knowledge written) had already happened
|
||||
// server-side. The agent's own work (a.chat below) still correctly stops
|
||||
// when ctx cancels — this only changes what happens to persistence.
|
||||
pctx := context.Background()
|
||||
|
||||
if sessionID == "" {
|
||||
title := truncate(req.Message, 80)
|
||||
sess, err := st.CreateSession(pctx, title)
|
||||
if err != nil {
|
||||
slog.Error("nomos: create session", "error", err)
|
||||
sessionID = "ephemeral"
|
||||
} else {
|
||||
sessionID = sess.ID
|
||||
}
|
||||
} else {
|
||||
// P2 iteration: if the operator sends a follow-up on a session
|
||||
// that already reached a terminal state (done/failed), reopen it
|
||||
// so a new sub-task can be framed (set_goal → propose_plan →
|
||||
// execute). reopenSession marks the prior plan's steps as
|
||||
// `replaced` (proposePlan ignores those) and clears outcome/
|
||||
// summary. Without this, propose_plan refuses the follow-up with
|
||||
// ErrPlanInFlight because the prior steps are all `done`. If the
|
||||
// session is still active, reopen is a no-op — the follow-up is
|
||||
// just a continuation of in-flight work.
|
||||
st.ReopenSession(pctx, sessionID)
|
||||
st.TouchSession(pctx, sessionID)
|
||||
}
|
||||
|
||||
slog.Info("nomos: chat", "session", sessionID, "message", truncate(req.Message, 100))
|
||||
|
||||
userMsg, _ := json.Marshal(map[string]any{"role": "user", "text": req.Message})
|
||||
st.SaveMessage(pctx, sessionID, "user", userMsg)
|
||||
|
||||
// If this task has a pending operator question, the incoming message IS the
|
||||
// answer — close it so the panel clears. No separate resume needed: this
|
||||
// chat turn is the resume, and the agent sees the question + answer in its
|
||||
// replayed history.
|
||||
if qid := st.OpenQuestionID(pctx, sessionID); qid != "" {
|
||||
st.AnswerQuestion(pctx, sessionID, qid, req.Message)
|
||||
}
|
||||
|
||||
writeEvent(agentEvent{Type: "session", Data: sessionID, SessionID: sessionID})
|
||||
|
||||
// F1/F2 (plan 2026-08-03): serialize turns per session. The user message is
|
||||
// already persisted above, so it is never lost. Wait briefly for a finishing
|
||||
// background turn; if one is still running after that, QUEUE this message
|
||||
// (don't reject it) and tell the client so it shows a "queued" state. The
|
||||
// in-flight turn's release drains the queue (drainQueued) and runs it as a
|
||||
// real turn server-side. This never stacks concurrent turns — the gate still
|
||||
// guarantees one in-flight turn per session.
|
||||
const turnWait = 5 * time.Second
|
||||
if !a.gate.Acquire(sessionID, turnWait) {
|
||||
a.queue.Enqueue(sessionID, req.Message)
|
||||
slog.Info("nomos: turn already active, queued operator message", "session", sessionID)
|
||||
writeEvent(agentEvent{Type: "queued", Data: sessionID, SessionID: sessionID})
|
||||
writeEvent(agentEvent{Type: "done", Data: map[string]any{
|
||||
"session_id": sessionID,
|
||||
"queued": true,
|
||||
}, SessionID: sessionID})
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
a.gate.Release(sessionID)
|
||||
// Run any message that was queued while this turn held the gate. In a
|
||||
// goroutine so the HTTP response finishes without waiting on the next
|
||||
// turn; the queued turn has no SSE client of its own.
|
||||
safego.Go("nomos:drain:"+sessionID, func() { a.drainQueued(context.Background(), sessionID) })
|
||||
}()
|
||||
|
||||
// F3 (plan 2026-08-03): keep the SSE alive during long turns. A turn can
|
||||
// run for many minutes (provisioning chains, deep research); the model
|
||||
// often takes 20-40s between tool iterations, and with nothing flushed in
|
||||
// that gap a proxy/browser idle timeout silently closes the stream. The
|
||||
// client then sees streaming=false while the server keeps working — the
|
||||
// "I can't tell it's working" desync. An SSE comment line (":keepalive") is
|
||||
// ignored by EventSource but resets idle timers.
|
||||
keepDone := make(chan struct{})
|
||||
go func() {
|
||||
t := time.NewTicker(12 * time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-keepDone:
|
||||
return
|
||||
case <-t.C:
|
||||
writeMu.Lock()
|
||||
fmt.Fprintf(w, ":keepalive\n\n")
|
||||
flusher.Flush()
|
||||
writeMu.Unlock()
|
||||
}
|
||||
}
|
||||
}()
|
||||
// Defer the close (not a statement after runChatTurn) so the goroutine
|
||||
// exits even if runChatTurn panics — net/http recovers handler panics, so
|
||||
// a non-deferred close would be skipped and the ticker would keep writing
|
||||
// to a dead ResponseWriter forever.
|
||||
defer close(keepDone)
|
||||
a.runChatTurn(pctx, ctx, sessionID, req.Message, func(ev agentEvent) {
|
||||
writeEvent(ev)
|
||||
})
|
||||
}
|
||||
|
||||
func handleSessionsList(w http.ResponseWriter, r *http.Request, st *session.Store) {
|
||||
if st == nil {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{"sessions": []any{}})
|
||||
return
|
||||
}
|
||||
|
||||
if r.Method == http.MethodOptions {
|
||||
return
|
||||
}
|
||||
|
||||
// P2.8 (2026-07-20): filtering + pagination. The audit script in
|
||||
// .agents/skills/session-review/SKILL.md slices `.sessions[:10]`
|
||||
// client-side; "show me partial sessions touching lxc:rclone"
|
||||
// required fetching the full list and filtering in JS. Push the
|
||||
// filters into SQL so the audit becomes a single `curl | jq`.
|
||||
// Supported query params (all optional, composable):
|
||||
// ?outcome=partial|success|failure — exact match on outcome
|
||||
// ?status=active|done|failed|executing — exact match on status
|
||||
// ?entity_id=<uuid> — exact match on entity_id
|
||||
// ?since=<RFC3339 or duration> — last_active_at >= ...
|
||||
// ?blocker=<reason> — exact match on blocker
|
||||
// ?limit=<int> — default 50, max 200
|
||||
// ?cursor=<iso timestamp> — last_active_at < cursor (page back)
|
||||
q := r.URL.Query()
|
||||
limit := 50
|
||||
if v := q.Get("limit"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 200 {
|
||||
limit = n
|
||||
}
|
||||
}
|
||||
sessions, err := st.ListSessionsFiltered(r.Context(), session.ListFilter{
|
||||
Outcome: q.Get("outcome"),
|
||||
Status: q.Get("status"),
|
||||
EntityID: q.Get("entity_id"),
|
||||
Blocker: q.Get("blocker"),
|
||||
Since: q.Get("since"),
|
||||
Cursor: q.Get("cursor"),
|
||||
Limit: limit,
|
||||
})
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
// Next-page cursor: the oldest last_active_at in this page. The next
|
||||
// request passes it as ?cursor=... to get the page before it. Empty
|
||||
// when the list is exhausted.
|
||||
var nextCursor string
|
||||
if len(sessions) > 0 {
|
||||
oldest := sessions[len(sessions)-1].LastActiveAt
|
||||
nextCursor = oldest.UTC().Format(time.RFC3339Nano)
|
||||
if len(sessions) < limit {
|
||||
nextCursor = "" // last page
|
||||
}
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"sessions": sessions,
|
||||
"next_cursor": nextCursor,
|
||||
"limit": limit,
|
||||
})
|
||||
}
|
||||
|
||||
func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *session.Store, a *agent) {
|
||||
if st == nil {
|
||||
http.Error(w, "not found", 404)
|
||||
return
|
||||
}
|
||||
|
||||
rest := strings.TrimPrefix(r.URL.Path, "/sessions/")
|
||||
parts := strings.Split(rest, "/")
|
||||
id := parts[0]
|
||||
if id == "" {
|
||||
http.Error(w, "session id required", 400)
|
||||
return
|
||||
}
|
||||
|
||||
// POST /sessions/{id}/questions/{qid}/answer — the operator answers a
|
||||
// pinned question from the context panel; resume the agent with the answer.
|
||||
if len(parts) == 4 && parts[1] == "questions" && parts[3] == "answer" {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
handleAnswerQuestion(w, r, st, a, id, parts[2])
|
||||
return
|
||||
}
|
||||
|
||||
// POST /sessions/{id}/resume — the operator asks the agent to continue.
|
||||
if len(parts) == 2 && parts[1] == "resume" && r.Method == http.MethodPost {
|
||||
base := "[System: the operator wants you to continue. Pick up where you left off — execute the next step of the plan, diagnose and fix any failures, or report progress if everything is done.]"
|
||||
note := st.EnrichResumeNote(context.Background(), id, base)
|
||||
safego.Go("nomos:resume-session", func() { a.resumeSession(context.Background(), id, note) })
|
||||
w.WriteHeader(202)
|
||||
return
|
||||
}
|
||||
|
||||
// GET /sessions/{id}/plan and /sessions/{id}/questions — REST hydration for
|
||||
// the context panel when it first opens a task; live events carry deltas
|
||||
// from there.
|
||||
// GET /sessions/{id}/tool_calls — flat view of every tool call in the
|
||||
// session, without the two-level message-shell nesting. The audit at
|
||||
// plans/2026-07-20-session-review-ten-sessions.md P2.10 had to write
|
||||
// Python to walk messages[].content.tool_calls[]; this endpoint makes
|
||||
// it a single `curl | jq`.
|
||||
if len(parts) == 2 && r.Method == http.MethodGet {
|
||||
switch parts[1] {
|
||||
case "plan":
|
||||
all := r.URL.Query().Has("all") && r.URL.Query().Get("all") != "0" && r.URL.Query().Get("all") != "false"
|
||||
steps, err := st.GetPlanSteps(r.Context(), id, all)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{"steps": steps})
|
||||
return
|
||||
case "questions":
|
||||
questions, err := st.GetQuestions(r.Context(), id)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{"questions": questions})
|
||||
return
|
||||
case "tool_calls":
|
||||
calls, err := st.GetSessionToolCalls(r.Context(), id)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{"session_id": id, "tool_calls": calls})
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
switch r.Method {
|
||||
case http.MethodDelete:
|
||||
if err := st.DeleteSession(r.Context(), id); err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(204)
|
||||
|
||||
case http.MethodGet:
|
||||
// P2.7 (2026-07-20): return BOTH session metadata and messages
|
||||
// from GET /sessions/{id}. Previously this endpoint returned only
|
||||
// {session_id, messages} — the operator had to merge with the
|
||||
// /sessions list view to get title/goal/outcome. The eval harness
|
||||
// at cmd/nomos/eval/main.go:302-303 already carries a comment
|
||||
// about this leaky abstraction. The session field carries the
|
||||
// full metadata: title, goal, outcome, summary, blocker,
|
||||
// pending_approvals, message_count, tool_call_count, etc. The
|
||||
// messages field is unchanged. Clients that only read
|
||||
// `messages` keep working.
|
||||
sess, err := st.GetSession(r.Context(), id)
|
||||
if err != nil {
|
||||
if err == pgx.ErrNoRows {
|
||||
http.Error(w, "session not found", 404)
|
||||
return
|
||||
}
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
messages, err := st.GetMessages(r.Context(), id)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"session_id": id,
|
||||
"session": sess,
|
||||
"messages": messages,
|
||||
})
|
||||
|
||||
default:
|
||||
http.Error(w, "method not allowed", 405)
|
||||
}
|
||||
}
|
||||
|
||||
// handleAnswerQuestion records the operator's answer to a pinned question and
|
||||
// resumes the agent in the background with that answer injected. Returns 202 —
|
||||
// the agent's response lands via the normal message-polling path, not this POST.
|
||||
func handleAnswerQuestion(w http.ResponseWriter, r *http.Request, st *session.Store, a *agent, sessionID, questionID string) {
|
||||
var req struct {
|
||||
Answer string `json:"answer"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || strings.TrimSpace(req.Answer) == "" {
|
||||
http.Error(w, "answer is required", 400)
|
||||
return
|
||||
}
|
||||
prompt, _, _ := st.GetQuestion(r.Context(), questionID)
|
||||
if err := st.AnswerQuestion(r.Context(), sessionID, questionID, req.Answer); err != nil {
|
||||
http.Error(w, err.Error(), 500)
|
||||
return
|
||||
}
|
||||
if a != nil {
|
||||
base := fmt.Sprintf("[System: the operator answered your question %q with: %q. "+
|
||||
"Continue the task from here — do not re-ask.]", prompt, req.Answer)
|
||||
note := st.EnrichResumeNote(context.Background(), sessionID, base)
|
||||
safego.Go("nomos:resume-session", func() { a.resumeSession(context.Background(), sessionID, note) })
|
||||
}
|
||||
w.WriteHeader(202)
|
||||
}
|
||||
|
||||
func handleQuery(w http.ResponseWriter, r *http.Request, pool *mcpClientPool, agentSlug, mcpURL string) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
|
||||
var req struct {
|
||||
Query string `json:"query"`
|
||||
Tool string `json:"tool"`
|
||||
Args map[string]any `json:"args"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "bad request: "+err.Error(), 400)
|
||||
return
|
||||
}
|
||||
|
||||
// The structured /query endpoint is stateless/session-less — "query" is a
|
||||
// fixed pool key (not a real session id) so repeated calls reuse one
|
||||
// dedicated connection instead of paying a fresh MCP handshake every time,
|
||||
// while still never sharing a connection with an actual chat task.
|
||||
client, err := pool.get("query")
|
||||
if err != nil {
|
||||
http.Error(w, "mcp unavailable: "+err.Error(), 502)
|
||||
return
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
|
||||
if req.Tool != "" {
|
||||
result, err := client.callTool(req.Tool, req.Args)
|
||||
duration := time.Since(start).Milliseconds()
|
||||
if err != nil {
|
||||
slog.Error("nomos: query failed", "tool", req.Tool, "error", err)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"error": err.Error(),
|
||||
"elapsed_ms": duration,
|
||||
"agent_slug": agentSlug,
|
||||
})
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"result": result,
|
||||
"elapsed_ms": duration,
|
||||
"agent_slug": agentSlug,
|
||||
"mcp_url": mcpURL,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
if req.Query != "" {
|
||||
if strings.Contains(strings.ToLower(req.Query), "what can you do") ||
|
||||
strings.Contains(strings.ToLower(req.Query), "help") {
|
||||
|
||||
tools, err := client.listTools()
|
||||
duration := time.Since(start).Milliseconds()
|
||||
if err != nil {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"error": err.Error(),
|
||||
"elapsed_ms": duration,
|
||||
})
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"message": "natural language queries belong to /chat. Use structured /query with 'tool' for direct MCP calls.",
|
||||
"tools": tools,
|
||||
"elapsed_ms": duration,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"message": "natural language queries belong to /chat. Use structured /query with 'tool' for direct MCP calls.",
|
||||
"elapsed_ms": time.Since(start).Milliseconds(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
http.Error(w, "either 'tool' or 'query' required", 400)
|
||||
}
|
||||
|
||||
func truncate(s string, n int) string {
|
||||
if len(s) <= n {
|
||||
return s
|
||||
}
|
||||
return s[:n] + "..."
|
||||
}
|
||||
@@ -1,510 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/dtoro/oikos/internal/nomos/session"
|
||||
)
|
||||
|
||||
// Task tools are nomos-LOCAL, not MCP tools. They are session-scoped, and the
|
||||
// shared MCP server (api:8090/mcp) has no session id — so these are handled
|
||||
// in-process by nomos, which knows the session/task and holds the store.
|
||||
// buildTools appends these to the model's tool list; the agent loop routes a
|
||||
// call whose name isTaskTool to handleTaskTool instead of the MCP client.
|
||||
//
|
||||
// Phase 3 ships complete_task; set_goal / propose_plan / update_plan_step /
|
||||
// ask_operator land in later phases through the same mechanism.
|
||||
|
||||
func taskToolDefs() []toolDef {
|
||||
return []toolDef{
|
||||
{
|
||||
Name: "set_goal",
|
||||
Description: "State the goal of this task in one sentence, as early as you " +
|
||||
"can. This is what the task is trying to achieve (e.g. 'Deploy TypeType " +
|
||||
"as an LXC on strong'); it heads the task on the board and the context " +
|
||||
"panel. Call it once you understand what the operator wants.",
|
||||
InputSchema: map[string]any{
|
||||
"type": "object",
|
||||
"properties": map[string]any{
|
||||
"goal": map[string]any{"type": "string", "description": "The task's goal, one sentence."},
|
||||
},
|
||||
"required": []string{"goal"},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "propose_plan",
|
||||
Description: "Propose the full ordered plan for this task. Call ONCE, before any " +
|
||||
"execution, with EVERY step end-to-end (not one step at a time). FIRST step: " +
|
||||
"research (prior knowledge, relations, blast radius). If your plan runs `run` " +
|
||||
"against any target, include a LAST step: write back " +
|
||||
"(update_entity_attributes + create_relationship + upsert_knowledge) — if you " +
|
||||
"omit it, one is auto-appended. After this call: STOP and wait for operator " +
|
||||
"approval (approval vocabulary: approved, yes, go, proceed, continue, ok, " +
|
||||
"go ahead). Once a step has started (running/done/...), this tool REFUSES " +
|
||||
"further calls — advance with update_plan_step + run instead. Re-propose only " +
|
||||
"if the operator explicitly asks you to revise the whole plan. complete_task " +
|
||||
"with outcome=success is REFUSED if you ran `run` but didn't call " +
|
||||
"update_entity_attributes/create_relationship — write back before completing.",
|
||||
InputSchema: map[string]any{
|
||||
"type": "object",
|
||||
"properties": map[string]any{
|
||||
"steps": map[string]any{
|
||||
"type": "array",
|
||||
"description": "Ordered steps, first to last.",
|
||||
"items": map[string]any{
|
||||
"type": "object",
|
||||
"properties": map[string]any{
|
||||
"title": map[string]any{"type": "string", "description": "Short imperative step title (e.g. 'Create the LXC')."},
|
||||
"detail": map[string]any{"type": "string", "description": "Optional one-line detail."},
|
||||
"target_slug": map[string]any{"type": "string", "description": "Optional entity slug this step acts on (e.g. lxc:typetype)."},
|
||||
},
|
||||
"required": []string{"title"},
|
||||
},
|
||||
},
|
||||
},
|
||||
"required": []string{"steps"},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "update_plan_step",
|
||||
Description: "Advance a plan step as you work it. Set status to 'running' when " +
|
||||
"you start it (pass execution_id if the step queued a gated action, so " +
|
||||
"the board can auto-close it when that finishes), then 'done' / 'failed' " +
|
||||
"/ 'skipped' / 'blocked' when it resolves. Keeps the operator's progress " +
|
||||
"view honest.",
|
||||
InputSchema: map[string]any{
|
||||
"type": "object",
|
||||
"properties": map[string]any{
|
||||
"seq": map[string]any{"type": "integer", "description": "1-based step number from propose_plan."},
|
||||
"status": map[string]any{"type": "string", "enum": []string{"running", "done", "failed", "skipped", "blocked"}, "description": "New status for the step."},
|
||||
"execution_id": map[string]any{"type": "string", "description": "Optional execution UUID this step is running, so it auto-closes on completion."},
|
||||
},
|
||||
"required": []string{"seq", "status"},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "ask_operator",
|
||||
Description: "Ask the operator a question when you hit a real decision only " +
|
||||
"they can make — an ambiguous target, a trade-off, missing information, " +
|
||||
"or a destructive choice not already approved. This pins a structured " +
|
||||
"question card in the context panel (with your options and the entities " +
|
||||
"involved) and PAUSES the task until they answer; their answer resumes " +
|
||||
"you automatically. Do NOT use it for things you can determine yourself " +
|
||||
"with tools — only for genuine decisions.",
|
||||
InputSchema: map[string]any{
|
||||
"type": "object",
|
||||
"properties": map[string]any{
|
||||
"prompt": map[string]any{"type": "string", "description": "The question, stated plainly."},
|
||||
"why": map[string]any{"type": "string", "description": "Why you're asking / what's at stake."},
|
||||
"options": map[string]any{
|
||||
"type": "array", "items": map[string]any{"type": "string"},
|
||||
"description": "The choices, if it's a pick-one decision.",
|
||||
},
|
||||
"context_entities": map[string]any{
|
||||
"type": "array", "items": map[string]any{"type": "string"},
|
||||
"description": "Entity slugs relevant to the decision (shown as chips).",
|
||||
},
|
||||
},
|
||||
"required": []string{"prompt"},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: "complete_task",
|
||||
Description: "Mark the current task finished. Call this once the goal is " +
|
||||
"verified done — or when you've genuinely failed or only partially " +
|
||||
"succeeded. Sets the task's outcome and a one-line summary shown on the " +
|
||||
"task board. Record what you learned with upsert_knowledge BEFORE " +
|
||||
"completing, so future tasks on the same entities benefit.",
|
||||
InputSchema: map[string]any{
|
||||
"type": "object",
|
||||
"properties": map[string]any{
|
||||
"outcome": map[string]any{
|
||||
"type": "string",
|
||||
"enum": []string{"success", "failure", "partial"},
|
||||
"description": "Did the task achieve its goal?",
|
||||
},
|
||||
"summary": map[string]any{
|
||||
"type": "string",
|
||||
"description": "One line describing the result (shown on the task card).",
|
||||
},
|
||||
},
|
||||
"required": []string{"outcome", "summary"},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// toInt coerces a JSON tool-arg number (float64 after unmarshal) to int.
|
||||
func toInt(v any) int {
|
||||
switch n := v.(type) {
|
||||
case float64:
|
||||
return int(n)
|
||||
case int:
|
||||
return n
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
// toStringSlice coerces a JSON tool-arg array to a non-empty []string.
|
||||
func toStringSlice(v any) []string {
|
||||
arr, ok := v.([]any)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(arr))
|
||||
for _, e := range arr {
|
||||
if s, ok := e.(string); ok && strings.TrimSpace(s) != "" {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// handleTaskTool executes a nomos-local task tool. Returns (result, true) if it
|
||||
// handled the call, or (nil, false) if name is not a local task tool (so the
|
||||
// caller forwards it to the MCP client).
|
||||
func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args map[string]any) (any, bool) {
|
||||
switch name {
|
||||
case "set_goal":
|
||||
goal, _ := args["goal"].(string)
|
||||
if strings.TrimSpace(goal) == "" {
|
||||
return "error: set_goal needs a goal", true
|
||||
}
|
||||
if err := a.store.SetGoal(ctx, sessionID, goal); err != nil {
|
||||
return fmt.Sprintf("error setting goal: %v", err), true
|
||||
}
|
||||
// P1: the plan window is NOT opened here. Opening it on set_goal
|
||||
// meant any config_mutation `run` auto-executed with zero operator
|
||||
// approval, before a plan was even proposed (let alone approved) —
|
||||
// a safety regression confirmed live in session d0d562e0. The
|
||||
// window is now opened only when the operator approves a plan
|
||||
// (chat-assent grant or explicit approval in agent.go), which is
|
||||
// what the SOUL.md "approve the plan, not each step" model actually
|
||||
// describes. set_goal records the goal + flips status to executing
|
||||
// and nothing more.
|
||||
response := "Goal set: " + goal + ". NEXT: pre-plan with read-only tools (search_knowledge, get_entity, list_lxcs, get_relations), then propose_plan (mandatory — even read-only tasks need a one-step plan; the run handler refuses without one). After propose_plan: if all steps are read-only, execute immediately (no approval needed). If any step is config_mutation/destructive, stop and wait for operator approval."
|
||||
// P1.3 (2026-07-20): surface prior partial/failed sessions for the
|
||||
// same problem so the agent can pick up the thread instead of
|
||||
// rediscovering it. Three rclone sessions (a51e2086, 8acea2e3,
|
||||
// cb8c8a4a) all bounced off the classifier because each new session
|
||||
// started from scratch. The agent gets a hint with the prior
|
||||
// goal + summary; if it looks related, search_knowledge or open
|
||||
// the prior session's transcript (GET /sessions/{id}) before
|
||||
// re-planning. See plans/2026-07-20-session-review-ten-sessions.md.
|
||||
prior, _ := a.store.RecentPartialSessions(ctx, sessionID, 24*time.Hour)
|
||||
if len(prior) > 0 {
|
||||
var b strings.Builder
|
||||
b.WriteString("\n\nNOTE — recent unfinished sessions (last 24h, outcome=partial/failed):")
|
||||
for i, p := range prior {
|
||||
if i >= 5 {
|
||||
b.WriteString(fmt.Sprintf("\n ...and %d more", len(prior)-5))
|
||||
break
|
||||
}
|
||||
sum := p.Summary
|
||||
if sum == "" {
|
||||
sum = "(no summary)"
|
||||
}
|
||||
if len(sum) > 200 {
|
||||
sum = sum[:200] + "..."
|
||||
}
|
||||
b.WriteString(fmt.Sprintf("\n - %s (sid %s, outcome=%s): %s",
|
||||
p.Goal, p.ID[:8], p.Outcome, sum))
|
||||
}
|
||||
b.WriteString("\nIf any of these looks like the same problem, search_knowledge for the prior investigation or read it via GET /sessions/{id} before re-planning — don't rediscover what was already learned.")
|
||||
response += b.String()
|
||||
}
|
||||
return response, true
|
||||
|
||||
case "propose_plan":
|
||||
raw, _ := args["steps"].([]any)
|
||||
var steps []session.PlanStepInput
|
||||
for _, r := range raw {
|
||||
m, ok := r.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
title, _ := m["title"].(string)
|
||||
if strings.TrimSpace(title) == "" {
|
||||
continue
|
||||
}
|
||||
detail, _ := m["detail"].(string)
|
||||
target, _ := m["target_slug"].(string)
|
||||
steps = append(steps, session.PlanStepInput{Title: title, Detail: detail, TargetSlug: target})
|
||||
}
|
||||
if len(steps) == 0 {
|
||||
return "error: propose_plan needs at least one step with a title", true
|
||||
}
|
||||
// D.2: auto-append a writeback step if the agent didn't include one.
|
||||
// The agent consistently writes vague last steps ("record findings")
|
||||
// and then skips update_entity_attributes entirely (the #1 cause of
|
||||
// knowledge-graph drift). Appending an explicit writeback step makes
|
||||
// the seq-order enforcement (5.6) require it to be completed last,
|
||||
// and D.1's complete_task gate enforces the actual calls. Together
|
||||
// they close the loop structurally — neither relies on the agent
|
||||
// reading SOUL.md. The match is broadened past the literal tool
|
||||
// names so a natural-language step ("Write back: update entity
|
||||
// attributes…") isn't doubled by an auto-appended duplicate (P1.2).
|
||||
hasWritebackStep := false
|
||||
for _, st := range steps {
|
||||
t := strings.ToLower(st.Title + " " + st.Detail)
|
||||
if strings.Contains(t, "update_entity_attributes") ||
|
||||
strings.Contains(t, "create_relationship") ||
|
||||
strings.Contains(t, "upsert_knowledge") ||
|
||||
strings.Contains(t, "write back") ||
|
||||
strings.Contains(t, "writeback") {
|
||||
hasWritebackStep = true
|
||||
break
|
||||
}
|
||||
}
|
||||
appendedNote := ""
|
||||
if !hasWritebackStep {
|
||||
steps = append(steps, session.PlanStepInput{
|
||||
Title: "Write back: update_entity_attributes + create_relationship + upsert_knowledge",
|
||||
Detail: "Call update_entity_attributes for every entity you ran against (versions, states, counts, timestamps). Call create_relationship for any edge you discovered. Then upsert_knowledge about the affected entities (pass `about` as an array).",
|
||||
})
|
||||
appendedNote = fmt.Sprintf(" (appended a writeback step — your plan didn't include one; step %d)", len(steps))
|
||||
}
|
||||
persisted, err := a.store.ProposePlan(ctx, sessionID, steps)
|
||||
if err != nil {
|
||||
if errors.Is(err, session.ErrPlanInFlight) {
|
||||
// The plan is already in flight — refuse the re-proposal.
|
||||
// The agent must advance the existing plan with
|
||||
// update_plan_step + run. This is the structural fix for
|
||||
// the "plan added twice" sidebar drift the operator
|
||||
// reported: instead of appending (which duplicated) or
|
||||
// wiping (which lost progress), we refuse and direct.
|
||||
return "Plan already in flight — refusing duplicate proposal. Steps exist and at least one has started (running/done/...). To advance: call update_plan_step(seq=K, status=\"running\") then run(...) for step K's target, then update_plan_step(seq=K, status=\"done\"). Do not call propose_plan again. Re-propose only if the operator explicitly asks you to revise the whole plan (the session is reopened on a follow-up — prior steps are marked `replaced` and a fresh generation is started), and say so in your reply before calling it.", true
|
||||
}
|
||||
return fmt.Sprintf("error proposing plan: %v", err), true
|
||||
}
|
||||
// The writeback step is now always present (D.2 auto-appends it if
|
||||
// the agent forgot), so the old advisory nudge is replaced by the
|
||||
// structural gate: D.1 refuses complete_task without the actual
|
||||
// update_entity_attributes/create_relationship calls. Enumerate the
|
||||
// step seqs so the model knows exactly which numbers to address with
|
||||
// update_plan_step (seq is 1-based within this plan — the addressing
|
||||
// key, not a global counter).
|
||||
var seqs strings.Builder
|
||||
for i, p := range persisted {
|
||||
if i > 0 {
|
||||
seqs.WriteString("; ")
|
||||
}
|
||||
title := fmt.Sprint(p["title"])
|
||||
fmt.Fprintf(&seqs, "%v=%s", p["seq"], title)
|
||||
}
|
||||
result := fmt.Sprintf("Plan set (%d steps): %s.%s Address them with update_plan_step(seq=N). If all steps are read-only, execute now — call update_plan_step(running) + run for each step, no approval needed. If any step is config_mutation/destructive, STOP and wait for operator approval (\"approved\", \"yes\", \"go\", \"proceed\", \"continue\", \"ok\", \"go ahead\"). Do not call propose_plan again.", len(persisted), seqs.String(), appendedNote)
|
||||
return result, true
|
||||
|
||||
case "update_plan_step":
|
||||
seq := toInt(args["seq"])
|
||||
status, _ := args["status"].(string)
|
||||
execID, _ := args["execution_id"].(string)
|
||||
if seq <= 0 || status == "" {
|
||||
return "error: update_plan_step needs seq (>=1) and status", true
|
||||
}
|
||||
reason, _ := args["replaced_reason"].(string)
|
||||
if err := a.store.UpdatePlanStep(ctx, sessionID, seq, status, execID, reason); err != nil {
|
||||
if errors.Is(err, session.ErrPlanStepNotFound) {
|
||||
// The seq doesn't address a step in the CURRENT plan — most
|
||||
// often a stale 1-based number the model carried across a
|
||||
// re-plan, or an out-of-range seq. seq is generation-relative
|
||||
// (1..N within the latest propose_plan), so a superseded
|
||||
// generation's row is never touched (P0.1 fix 3). Direct the
|
||||
// model instead of silently no-op'ing.
|
||||
return fmt.Sprintf("Step %d is not in the current plan. seq is 1-based within your latest propose_plan (a re-plan resets it to 1..N, so an old step number no longer applies). The plan was not changed. Re-address with the correct 1-based seq, or if you've lost track, re-read the plan.", seq), true
|
||||
}
|
||||
return fmt.Sprintf("error updating step %d: %v", seq, err), true
|
||||
}
|
||||
return fmt.Sprintf("Step %d → %s. (Advance with update_plan_step + run; do not re-propose.)", seq, status), true
|
||||
|
||||
case "ask_operator":
|
||||
prompt, _ := args["prompt"].(string)
|
||||
if strings.TrimSpace(prompt) == "" {
|
||||
return "error: ask_operator needs a prompt", true
|
||||
}
|
||||
qctx := map[string]any{}
|
||||
if why, _ := args["why"].(string); strings.TrimSpace(why) != "" {
|
||||
qctx["why"] = why
|
||||
}
|
||||
if opts := toStringSlice(args["options"]); len(opts) > 0 {
|
||||
qctx["options"] = opts
|
||||
}
|
||||
if ents := toStringSlice(args["context_entities"]); len(ents) > 0 {
|
||||
qctx["entities"] = ents
|
||||
}
|
||||
if _, err := a.store.AskOperator(ctx, sessionID, prompt, qctx); err != nil {
|
||||
return fmt.Sprintf("error posting question: %v", err), true
|
||||
}
|
||||
return "Question posted to the operator; the task is paused until they answer. " +
|
||||
"Do not continue or call more tools — end your turn now and wait for their answer.", true
|
||||
|
||||
case "complete_task":
|
||||
outcome, _ := args["outcome"].(string)
|
||||
summary, _ := args["summary"].(string)
|
||||
switch outcome {
|
||||
case "":
|
||||
outcome = "success" // no outcome given at all — assume success, the common case
|
||||
case "success", "failure", "partial":
|
||||
// valid, use as-is
|
||||
default:
|
||||
// The tool schema declares an enum, but a weaker model (or a
|
||||
// typo) can still send anything — an unrecognized value used to
|
||||
// persist as-is, silently, with only "failure" special-cased
|
||||
// (store.completeTask derives status='failed' from it; anything
|
||||
// else became status='done' regardless of what the value
|
||||
// actually said). Default to "partial" rather than silently
|
||||
// treating an unrecognized value as "success" — safer to
|
||||
// under-claim than over-claim a task's outcome.
|
||||
slog.Warn("nomos: complete_task got an unrecognized outcome, defaulting to partial",
|
||||
"session", sessionID, "outcome", outcome)
|
||||
outcome = "partial"
|
||||
}
|
||||
// D.1: refuse success when discovery ran but no writeback followed.
|
||||
// The prior advisory warning (below) was ignorable — the agent
|
||||
// saw it and ended the task anyway. This gate fires BEFORE
|
||||
// completeTask runs, so the session stays in 'executing' state
|
||||
// and the agent must call update_entity_attributes/create_relationship
|
||||
// then retry complete_task. Only blocks `success`; an explicit
|
||||
// `failure` or `partial` is allowed through (the agent is
|
||||
// acknowledging it didn't finish — no reason to force writeback).
|
||||
if outcome == "success" && a.store.HadDiscovery(ctx, sessionID) && !a.store.HadEntityWriteback(ctx, sessionID) {
|
||||
return "Refused: this session ran `run` against live targets (discovery) but did not call update_entity_attributes or create_relationship to persist what you learned. The knowledge graph will drift if you complete without writeback. Call update_entity_attributes for each entity you ran against (versions, states, counts, timestamps), and create_relationship for any edge you discovered, then call complete_task again. Outcome is held at 'executing' until you do.", true
|
||||
}
|
||||
// D.2: refuse success when the goal mentions a reachability/uptime
|
||||
// check but no verification was done. The agent can't claim "X is
|
||||
// reachable" based on a shell command alone — the proxy (Caddy) can
|
||||
// return 200 for a terminal page (ttyd) or fallback while the actual
|
||||
// dashboard is still down. Must call ping_service or run a successful
|
||||
// curl before claiming success.
|
||||
if outcome == "success" && a.store.HadDiscovery(ctx, sessionID) {
|
||||
goal := a.store.SessionGoal(ctx, sessionID)
|
||||
if mentionsReachability(goal) && !a.store.HadRecentVerification(ctx, sessionID) {
|
||||
return "Refused: the goal involves a reachability or uptime check (\"make X reachable\", \"get X up\", etc.), but no ping_service call or successful curl/HTTP request against the target was detected. Caddy can return 200 for a terminal or fallback page while the actual service is still down — you must verify the service itself, not just the proxy. Call ping_service(target) or run a curl against the actual service URL, then call complete_task again. Outcome held until verified.", true
|
||||
}
|
||||
}
|
||||
if err := a.store.CompleteTask(ctx, sessionID, outcome, summary); err != nil {
|
||||
if errors.Is(err, session.ErrTaskAlreadyComplete) {
|
||||
return "Task is already complete. Do not call complete_task again. If the operator pointed out a UI/sidebar inconsistency, fix it with update_plan_step (reconcile step states) or summarize the panel in your reply — do not re-execute the work.", true
|
||||
}
|
||||
return fmt.Sprintf("error completing task: %v", err), true
|
||||
}
|
||||
result := fmt.Sprintf("Task marked %s: %s", outcome, summary)
|
||||
if !a.store.HadEntityWriteback(ctx, sessionID) {
|
||||
result += "\n\n⚠️ No entity attributes or relationships were updated in this session. Call update_entity_attributes and create_relationship to persist what you learned about entities before the next session starts from scratch."
|
||||
}
|
||||
return result, true
|
||||
default:
|
||||
return nil, false
|
||||
}
|
||||
}
|
||||
|
||||
// reachabilityPatterns matches goal text that involves making something
|
||||
// reachable/accessible/working. Used by complete_task to surface a soft
|
||||
// warning when the session goal was about reachability but no verification
|
||||
// occurred before marking success.
|
||||
var reachabilityPatterns = []*regexp.Regexp{
|
||||
regexp.MustCompile(`(?i)https?://[^\s]+`),
|
||||
regexp.MustCompile(`(?i)\.hubris\.net\w+`),
|
||||
regexp.MustCompile(`(?i)(un)?reachable`),
|
||||
regexp.MustCompile(`(?i)(not?\s+)?(accessible|reachable|responding|resolving)`),
|
||||
regexp.MustCompile(`(?i)diagnose\s+why`),
|
||||
regexp.MustCompile(`(?i)(fix|restore|bring\s+back).*(accessible|reachable|online)`),
|
||||
}
|
||||
|
||||
func mentionsReachability(goal string) bool {
|
||||
for _, p := range reachabilityPatterns {
|
||||
if p.MatchString(goal) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// autoCompleteTrivialTask is the case-1 fix from
|
||||
// plans/2026-07-11-task-completion-safety-net.md: a session that never
|
||||
// called set_goal never framed itself as a structured task, so a turn that
|
||||
// ends with a plain-text answer and no further tool calls IS the task
|
||||
// ending — but the model consistently skips complete_task for exactly this
|
||||
// case (confirmed live: 43/50 production sessions were a single trivial
|
||||
// Q&A exchange, none of which ever reached a terminal status). Rather than
|
||||
// leave agent_sessions.status stuck at its creation-time default forever,
|
||||
// close it out mechanically here: no judgment call needed, since SOUL.md
|
||||
// already treats a one-shot answered question as done by definition.
|
||||
func (a *agent) autoCompleteTrivialTask(ctx context.Context, sessionID, responseText string) {
|
||||
summary := strings.TrimSpace(responseText)
|
||||
summary = strings.SplitN(summary, "\n", 2)[0] // first line only — the board shows one line
|
||||
const maxLen = 120
|
||||
if len(summary) > maxLen {
|
||||
summary = summary[:maxLen] + "…"
|
||||
}
|
||||
if summary == "" {
|
||||
summary = "Answered without further action needed."
|
||||
}
|
||||
if err := a.store.CompleteTask(ctx, sessionID, "success", summary); err != nil {
|
||||
slog.Error("nomos: auto-complete trivial task failed", "session", sessionID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// autoCompleteIfPlanDone is the structural safety net for "the agent did the
|
||||
// work but forgot to call complete_task" — the #1 remaining model reliability
|
||||
// gap after D.1's writeback gate. After a turn ends, if the session has a goal,
|
||||
// the agent never called complete_task this turn, and either (a) all plan
|
||||
// steps are terminal OR (b) the agent did discovery (ran `run`), auto-complete.
|
||||
// Path (b) catches the common case where the agent skips update_plan_step
|
||||
// bookkeeping but still does the actual work — the D.1 gate already enforces
|
||||
// writeback before `complete_task`, so if the agent forgot to complete at all,
|
||||
// we close it out mechanically. If writeback happened → success; if not →
|
||||
// partial (honest: work was done but knowledge graph wasn't updated).
|
||||
func (a *agent) autoCompleteIfPlanDone(ctx context.Context, sessionID, responseText string) {
|
||||
if a.store == nil || sessionID == "" || sessionID == "ephemeral" {
|
||||
return
|
||||
}
|
||||
sess, err := a.store.GetSession(ctx, sessionID)
|
||||
if err != nil || sess.Status != "executing" {
|
||||
return
|
||||
}
|
||||
// Don't auto-complete if there are pending approvals — the agent is
|
||||
// blocked waiting for the operator, not done. Auto-completing here
|
||||
// would close the session and the operator's approval would land on a
|
||||
// dead task. Confirmed in eval: agent hits P5 approval gate, turn
|
||||
// ends, auto-complete fires incorrectly because the approval-queue
|
||||
// `run` responses were logged as success=true in agent_activity.
|
||||
if a.store.HasPendingApprovals(ctx, sessionID) {
|
||||
return
|
||||
}
|
||||
discovery := a.store.HadDiscovery(ctx, sessionID)
|
||||
writeback := a.store.HadEntityWriteback(ctx, sessionID)
|
||||
// (a) all plan steps terminal, OR (b) agent did discovery (ran `run`).
|
||||
shouldComplete := a.store.AllPlanStepsTerminal(ctx, sessionID)
|
||||
if !shouldComplete && discovery {
|
||||
shouldComplete = true
|
||||
}
|
||||
if !shouldComplete {
|
||||
return
|
||||
}
|
||||
outcome := "success"
|
||||
if discovery && !writeback {
|
||||
outcome = "partial" // honest: work done, knowledge graph not updated
|
||||
}
|
||||
summary := strings.TrimSpace(responseText)
|
||||
summary = strings.SplitN(summary, "\n", 2)[0]
|
||||
const maxLen = 120
|
||||
if len(summary) > maxLen {
|
||||
summary = summary[:maxLen] + "…"
|
||||
}
|
||||
if summary == "" {
|
||||
summary = "All plan steps completed."
|
||||
}
|
||||
if err := a.store.CompleteTask(ctx, sessionID, outcome, summary); err != nil {
|
||||
slog.Error("nomos: auto-complete plan-done task failed", "session", sessionID, "error", err)
|
||||
} else {
|
||||
slog.Info("nomos: auto-completed task — agent didn't call complete_task", "session", sessionID, "outcome", outcome)
|
||||
}
|
||||
}
|
||||
@@ -1,152 +0,0 @@
|
||||
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) {})
|
||||
}()
|
||||
}
|
||||
}
|
||||
@@ -150,7 +150,7 @@ Roles:
|
||||
knowledge Convert wiki to knowledge seed (one-shot)
|
||||
version Print version info
|
||||
|
||||
The operator interface is Nomos (MCP agent) — no CLI needed.
|
||||
The operator interface is dsh (MCP agent) — no CLI needed.
|
||||
Environment:
|
||||
OIKOS_DATABASE_URL Postgres connection string
|
||||
OIKOS_API_LISTEN API listen address (default :8090)
|
||||
|
||||
@@ -17,21 +17,84 @@ import (
|
||||
"github.com/dtoro/oikos/internal/safego"
|
||||
)
|
||||
|
||||
type webhookHandler struct {
|
||||
secret string
|
||||
script string
|
||||
}
|
||||
|
||||
func (h *webhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
http.Error(w, "read body failed", 400)
|
||||
return
|
||||
}
|
||||
|
||||
sigHex := r.Header.Get("X-Hub-Signature-256")
|
||||
if sigHex == "" {
|
||||
http.Error(w, "missing signature", 401)
|
||||
return
|
||||
}
|
||||
|
||||
mac := hmac.New(sha256.New, []byte(h.secret))
|
||||
mac.Write(body)
|
||||
expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))
|
||||
|
||||
if !hmac.Equal([]byte(sigHex), []byte(expected)) {
|
||||
slog.Warn("webhook: invalid signature")
|
||||
http.Error(w, "invalid signature", 401)
|
||||
return
|
||||
}
|
||||
|
||||
slog.Info("webhook: deploy triggered", "script", h.script)
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
w.Write([]byte(`{"status":"deploy started"}`))
|
||||
|
||||
safego.Go("webhook:"+h.script, func() {
|
||||
apiToken := ""
|
||||
if sec != nil {
|
||||
apiToken = secrets.ResolveSecret(ctx, sec, "api_token", "")
|
||||
}
|
||||
cmd := exec.Command(h.script)
|
||||
cmd.Dir = repoDir
|
||||
cmd.Env = append(os.Environ(),
|
||||
"REPO_DIR="+repoDir,
|
||||
"PROFILE=full",
|
||||
"OIKOS_API_TOKEN="+apiToken,
|
||||
)
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
start := time.Now()
|
||||
if err := cmd.Run(); err != nil {
|
||||
slog.Error("webhook: deploy failed", "script", h.script, "error", err, "duration", time.Since(start))
|
||||
return
|
||||
}
|
||||
slog.Info("webhook: deploy succeeded", "script", h.script, "duration", time.Since(start))
|
||||
})
|
||||
}
|
||||
|
||||
var ctx context.Context
|
||||
var sec *secrets.Manager
|
||||
var repoDir string
|
||||
|
||||
func main() {
|
||||
ctx := context.Background()
|
||||
ctx = context.Background()
|
||||
|
||||
port := os.Getenv("WEBHOOK_LISTEN")
|
||||
if port == "" {
|
||||
port = ":9797"
|
||||
}
|
||||
|
||||
repoDir := os.Getenv("WEBHOOK_REPO_DIR")
|
||||
repoDir = os.Getenv("WEBHOOK_REPO_DIR")
|
||||
if repoDir == "" {
|
||||
repoDir = os.Getenv("HOME") + "/Projects/oikos"
|
||||
}
|
||||
|
||||
// Create secrets manager once, share between HMAC resolution and deploy
|
||||
sec := newSecrets()
|
||||
sec = newSecrets()
|
||||
secret := resolveWebhookHMAC(ctx, sec)
|
||||
if secret == "" {
|
||||
fmt.Fprintln(os.Stderr, "WEBHOOK_HMAC_SECRET must be set (env var or Infisical webhook_hmac-secret)")
|
||||
@@ -39,60 +102,8 @@ func main() {
|
||||
}
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/deploy", func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
http.Error(w, "read body failed", 400)
|
||||
return
|
||||
}
|
||||
|
||||
sigHex := r.Header.Get("X-Hub-Signature-256")
|
||||
if sigHex == "" {
|
||||
http.Error(w, "missing signature", 401)
|
||||
return
|
||||
}
|
||||
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
mac.Write(body)
|
||||
expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))
|
||||
|
||||
if !hmac.Equal([]byte(sigHex), []byte(expected)) {
|
||||
slog.Warn("webhook: invalid signature")
|
||||
http.Error(w, "invalid signature", 401)
|
||||
return
|
||||
}
|
||||
|
||||
slog.Info("webhook: deploy triggered")
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
w.Write([]byte(`{"status":"deploy started"}`))
|
||||
|
||||
safego.Go("webhook:deploy", func() {
|
||||
apiToken := ""
|
||||
if sec != nil {
|
||||
apiToken = secrets.ResolveSecret(ctx, sec, "api_token", "")
|
||||
}
|
||||
cmd := exec.Command(repoDir + "/scripts/deploy.sh")
|
||||
cmd.Dir = repoDir
|
||||
cmd.Env = append(os.Environ(),
|
||||
"REPO_DIR="+repoDir,
|
||||
"PROFILE=full",
|
||||
"OIKOS_API_TOKEN="+apiToken,
|
||||
)
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
start := time.Now()
|
||||
if err := cmd.Run(); err != nil {
|
||||
slog.Error("webhook: deploy failed", "error", err, "duration", time.Since(start))
|
||||
return
|
||||
}
|
||||
slog.Info("webhook: deploy succeeded", "duration", time.Since(start))
|
||||
})
|
||||
})
|
||||
mux.Handle("/deploy", &webhookHandler{secret: secret, script: repoDir + "/scripts/deploy.sh"})
|
||||
mux.Handle("/deploy-plugins", &webhookHandler{secret: secret, script: repoDir + "/scripts/deploy-plugins.sh"})
|
||||
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(200)
|
||||
@@ -106,7 +117,6 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
// newSecrets creates the Infisical secrets manager from env vars.
|
||||
func newSecrets() *secrets.Manager {
|
||||
return secrets.NewManagerFromConfig(
|
||||
os.Getenv("OIKOS_INFISICAL_SITE_URL"),
|
||||
@@ -118,8 +128,6 @@ func newSecrets() *secrets.Manager {
|
||||
)
|
||||
}
|
||||
|
||||
// resolveWebhookHMAC fetches the webhook HMAC secret from Infisical,
|
||||
// falling back to the WEBHOOK_HMAC_SECRET env var.
|
||||
func resolveWebhookHMAC(ctx context.Context, sec *secrets.Manager) string {
|
||||
envFallback := os.Getenv("WEBHOOK_HMAC_SECRET")
|
||||
if sec == nil {
|
||||
|
||||
Reference in New Issue
Block a user