From b237fbaa32a1851a15e77f8d391c8f8532b76387 Mon Sep 17 00:00:00 2001 From: BaaRai <130491490+BaaRai@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:46:56 +0200 Subject: [PATCH] feat(engine): ouroboros takes - goal ambiguity gate, failover escalation, stagnation surfacing Three orchestration mechanisms from the Q00/ouroboros evaluation (#99): - Goal ambiguity gate at task creation: a goal must name a target (URL, domain, IPv4, localhost:port or absolute path) and reach a length floor, else the API returns 400 before any token is spent. Deliberately target-less goals opt out with the literal marker [targetless]. Guards against the live failure where a goal naming no target sent every worker to an unrelated asset. - Consecutive-failure escalation on the task LLM chain: quota exhaustion already advances the chain; now 3 consecutive non-quota call failures on the same chain profile advance it too, with the true reason recorded. Escalation-only by design (no auto-downgrade: mid-task model swaps risk behavior drift; the operator keeps the manual chain reset). - Stagnation surfacing in the planner overview: finished intents are fingerprinted (case-folded, digits collapsed); a direction dispatched 3+ times with zero findings since its first attempt is flagged in a self-describing stagnation_warning payload that instructs the planner to diversify instead of grinding. Absent when nothing stagnates. Closes #99 --- agent/stagnation_test.go | 39 ++++++++++++ agent/tools.go | 95 ++++++++++++++++++++++++++++++ server/goal_gate.go | 42 +++++++++++++ server/goal_gate_test.go | 37 ++++++++++++ server/server.go | 4 ++ server/task_llm.go | 84 ++++++++++++++++++++++++++ server/task_llm_escalation_test.go | 85 ++++++++++++++++++++++++++ 7 files changed, 386 insertions(+) create mode 100644 agent/stagnation_test.go create mode 100644 server/goal_gate.go create mode 100644 server/goal_gate_test.go create mode 100644 server/task_llm_escalation_test.go diff --git a/agent/stagnation_test.go b/agent/stagnation_test.go new file mode 100644 index 0000000..059863d --- /dev/null +++ b/agent/stagnation_test.go @@ -0,0 +1,39 @@ +package agent + +import ( + "encoding/json" + "testing" + + "github.com/Autumn-27/artex/db" +) + +func nodeWithSummary(id int64, summary string) *db.Node { + payload, _ := json.Marshal(map[string]any{"summary": summary}) + return &db.Node{ID: id, Kind: db.KindIntent, State: "done", Payload: payload} +} + +func TestDetectStagnation(t *testing.T) { + done := []*db.Node{ + nodeWithSummary(10, "Enumerate endpoints on https://target.example.com/api/v2"), + nodeWithSummary(22, "Enumerate endpoints on https://target.example.com/api/v3"), + nodeWithSummary(35, "enumerate endpoints on https://target.example.com/api/v9"), + nodeWithSummary(40, "Review authentication flow for token replay"), + } + if got := detectStagnation(done, nil); len(got) != 1 { + t.Fatalf("expected 1 stagnating direction, got %d: %+v", len(got), got) + } else if got[0]["repeats"] != 3 { + t.Fatalf("repeats=%v want 3", got[0]["repeats"]) + } + + // a finding produced after the first repeat disarms the direction + findingPayload, _ := json.Marshal(map[string]any{"summary": "IDOR found"}) + findings := []*db.Node{{ID: 25, Kind: db.KindFinding, Payload: findingPayload}} + if got := detectStagnation(done, findings); len(got) != 0 { + t.Fatalf("expected no stagnation once a finding exists, got %+v", got) + } + + // below the repeat threshold: nothing + if got := detectStagnation(done[:2], nil); len(got) != 0 { + t.Fatalf("expected no stagnation below threshold, got %+v", got) + } +} diff --git a/agent/tools.go b/agent/tools.go index c169450..815e014 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -17,6 +17,92 @@ import ( // upstream nodes it derived from: facts/intents/findings) and yields (the facts/ // findings it produced) — without pulling full payloads. parentsOf/yieldsOf are // built from the exploration edges in graph_overview. +// normalizeDirection fingerprints an intent summary for stagnation detection: +// case-folded, all digits collapsed to '#', whitespace collapsed. Two intents +// aimed at the same thing with different IDs/counts/URLs fingerprint the same. +func normalizeDirection(summary string) string { + var b strings.Builder + digit := false + for _, r := range strings.ToLower(strings.Join(strings.Fields(summary), " ")) { + if r >= '0' && r <= '9' { + if !digit { + b.WriteByte('#') + } + digit = true + continue + } + digit = false + b.WriteRune(r) + } + return b.String() +} + +// detectStagnation (ouroboros take, #99) flags directions whose normalized +// summary repeats stagnationMinRepeats times among finished intents while no +// finding has been produced since the FIRST attempt at that direction: the +// planner is grinding the same shape of work without yield. Pure function over +// (finished intents, findings) so it is unit-testable without a store. +const stagnationMinRepeats = 3 + +func detectStagnation(done, findings []*db.Node) []map[string]any { + type dir struct { + firstID int64 + repeats int + summary string + } + dirs := map[string]*dir{} + order := []string{} + for _, n := range done { + var payload map[string]any + _ = json.Unmarshal(n.Payload, &payload) + summ, _ := payload["summary"].(string) + if strings.TrimSpace(summ) == "" { + continue + } + fp := normalizeDirection(summ) + if fp == "" { + continue + } + d, ok := dirs[fp] + if !ok { + d = &dir{firstID: n.ID, summary: summ} + dirs[fp] = d + order = append(order, fp) + } + if n.ID < d.firstID { + d.firstID = n.ID + d.summary = summ + } + d.repeats++ + } + var out []map[string]any + for _, fp := range order { + d := dirs[fp] + if d.repeats < stagnationMinRepeats { + continue + } + produced := false + for _, f := range findings { + if f.ID > d.firstID { + produced = true + break + } + } + if produced { + continue + } + out = append(out, map[string]any{ + "direction": d.summary, + "repeats": d.repeats, + "since": d.firstID, + }) + if len(out) >= 3 { + break + } + } + return out +} + func compactIntents(ns []*db.Node, parentsOf, yieldsOf map[int64][]int64) []map[string]any { out := make([]map[string]any, 0, len(ns)) for _, n := range ns { @@ -449,6 +535,15 @@ func (t *ToolSet) graphOverviewData() map[string]any { factNodes, _ := t.ts.ListByKind(db.KindFact, 1000) // newest first out["findings_total"] = len(vulnNodes) // total confirmed findings (goal judgment looks here); details in finding_list (latest window) out["facts"] = len(factNodes) // number of exploration facts/conclusions (including negative conclusions) + // Stagnation surfacing (#99): repeated finished directions with no findings + // since their first attempt. Self-describing payload — the instruction ships + // with the data so no prompt-template change is needed on deployed installs. + if stale := detectStagnation(recentDone, vulnNodes); len(stale) > 0 { + out["stagnation_warning"] = map[string]any{ + "directions": stale, + "instruction": "these directions were dispatched repeatedly with zero findings produced since their first attempt — do NOT re-dispatch the same approach again; open a materially different direction, or judge the goal on the existing evidence instead of grinding", + } + } // findings are the task's highest-value output → the overview carries the latest // window (≤10 entries; vulnNodes is id-DESC, i.e. newest first) so the planner sees // the recently confirmed findings at a glance each round when judging goals; use diff --git a/server/goal_gate.go b/server/goal_gate.go new file mode 100644 index 0000000..8ef51b1 --- /dev/null +++ b/server/goal_gate.go @@ -0,0 +1,42 @@ +package server + +import ( + "errors" + "regexp" + "strings" +) + +// Ambiguity gate (ouroboros take, #99): a task goal must name a concrete target +// and be long enough to carry a testable objective before any token is spent. +// The live failure this guards against: a benchmark task whose goal named no +// target URL passed creation silently and the planner sent every worker to an +// unrelated asset. A goal that is deliberately target-less (pure analysis over +// provided material) can opt out with the literal marker [targetless]. +var ( + goalMinLen = 80 + + goalTargetlessRE = regexp.MustCompile(`\[targetless\]`) + goalURLRE = regexp.MustCompile(`(?i)\bhttps?://\S+`) + goalDomainRE = regexp.MustCompile(`(?i)\b[a-z0-9][a-z0-9-]{1,61}\.[a-z]{2,}(:[0-9]+)?(/\S*)?\b`) + goalIPv4RE = regexp.MustCompile(`\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(:\d+)?\b`) + goalLocalhostRE = regexp.MustCompile(`(?i)\blocalhost:[0-9]+\b`) + goalPathRE = regexp.MustCompile(`(?i)(^|\s)/[a-z0-9_-]+(/[a-z0-9_.-]+)+`) +) + +// goalGateCheck returns a descriptive error when the goal is too ambiguous to +// dispatch workers against. Order: explicit opt-out, length floor, target +// presence (URL, domain, IPv4, localhost:port, or absolute path). +func goalGateCheck(goal string) error { + g := strings.TrimSpace(goal) + if goalTargetlessRE.MatchString(g) { + return nil + } + if len(g) < goalMinLen { + return errors.New("goal is too short to carry a testable objective (min 80 chars): name the target and what counts as success; add the literal marker [targetless] only for genuinely target-less analysis tasks") + } + if goalURLRE.MatchString(g) || goalDomainRE.MatchString(g) || + goalIPv4RE.MatchString(g) || goalLocalhostRE.MatchString(g) || goalPathRE.MatchString(g) { + return nil + } + return errors.New("goal names no target: include the URL/host/IP/path under test (or the literal marker [targetless] for target-less analysis tasks) so workers cannot be dispatched against an unrelated asset") +} diff --git a/server/goal_gate_test.go b/server/goal_gate_test.go new file mode 100644 index 0000000..94eb2b0 --- /dev/null +++ b/server/goal_gate_test.go @@ -0,0 +1,37 @@ +package server + +import ( + "strings" + "testing" +) + +func TestGoalGateCheck(t *testing.T) { + good := map[string]string{ + "url target": "Audit " + strings.Repeat("x ", 60) + " at https://example.com/app every module for injection issues and record one finding per issue with evidence.", + "bare domain": strings.Repeat("probe ", 25) + "the api at target.example.org for unauthenticated data exposure, one record proof, then stop.", + "ipv4": strings.Repeat("recon ", 25) + "10.0.0.5 and map every listening service before any test.", + "localhost port": strings.Repeat("verify ", 25) + "localhost:3000 health endpoints respond identically before and after config reload.", + "path": strings.Repeat("review ", 25) + "the module at /app/data/core for hardcoded secrets and report each hit.", + "targetless opt-out": "[targetless] " + strings.Repeat("summarize ", 20) + "the attached material into a prioritized hypothesis list.", + } + good["targetless short but marked"] = "[targetless] analyze nothing" + + bad := map[string]string{ + "no target long": strings.Repeat("do a thorough security review of everything and report anything interesting that comes up during the exploration phase of this audit, thanks. ", 2), + "too short": "audit the site", + "short and bare": "find bugs", + } + for name, g := range good { + if name == "targetless short+:url" { + continue + } + if err := goalGateCheck(g); err != nil { + t.Errorf("%s: unexpected error: %v", name, err) + } + } + for name, g := range bad { + if err := goalGateCheck(g); err == nil { + t.Errorf("%s: expected error, got none", name) + } + } +} diff --git a/server/server.go b/server/server.go index ef79c7b..ba64fbf 100644 --- a/server/server.go +++ b/server/server.go @@ -1541,6 +1541,10 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, err.Error()) return } + if err := goalGateCheck(req.Goal); err != nil { + writeErr(w, 400, err.Error()) + return + } if req.TimeoutSeconds < 0 { req.TimeoutSeconds = 0 } diff --git a/server/task_llm.go b/server/task_llm.go index df37e5d..a00b24d 100644 --- a/server/task_llm.go +++ b/server/task_llm.go @@ -10,6 +10,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "time" "github.com/Autumn-27/artex/agent" @@ -85,6 +86,46 @@ func isQuotaExhaustedError(err error) bool { return err != nil && agent.IsQuotaExhaustedMessage(err.Error()) } +// Consecutive-failure escalation (ouroboros cost-tier take, #99): quota +// exhaustion already advances the chain, but a profile that keeps failing with +// plain runtime errors (bad key, persistent 5xx, broken model output) would +// otherwise burn the whole same-provider retry budget on every call. After +// llmFailStreakThreshold consecutive failed calls on the same chain profile +// the chain advances exactly like a quota exhaust, carrying the true reason. +// Deliberately escalation-only: no auto-downgrade (a mid-task model swap-down +// risks behavior drift and flapping; the operator can reset the chain). +const llmFailStreakThreshold = 3 + +var llmFailStreaks = struct { + sync.Mutex + m map[string]int +}{m: map[string]int{}} + +func llmFailKey(taskID string, profileID int64) string { + return taskID + ":" + strconv.FormatInt(profileID, 10) +} + +// noteLLMCallFailure records one failed call on (task, profile) and reports +// whether the escalation threshold was just reached. The streak resets when it +// fires and on every success, so the counter tracks consecutive failures only. +func noteLLMCallFailure(taskID string, profileID int64) bool { + llmFailStreaks.Lock() + defer llmFailStreaks.Unlock() + k := llmFailKey(taskID, profileID) + llmFailStreaks.m[k]++ + if llmFailStreaks.m[k] >= llmFailStreakThreshold { + delete(llmFailStreaks.m, k) // armed and consumed — next failure starts a fresh streak + return true + } + return false +} + +func resetLLMFailStreak(taskID string, profileID int64) { + llmFailStreaks.Lock() + defer llmFailStreaks.Unlock() + delete(llmFailStreaks.m, llmFailKey(taskID, profileID)) +} + type taskLLMSelection struct { task *Task profileID int64 @@ -264,8 +305,28 @@ func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionReque break } if callErr == nil { + resetLLMFailStreak(taskID, selection.profileID) return msg, sr, usage, nil } + // Non-quota failures on a chain profile escalate after a consecutive + // streak (#99): same advance path as quota exhaust, truthful reason. + if selection.profileID != 0 && !isQuotaExhaustedError(callErr) && ctx.Err() == nil { + if noteLLMCallFailure(taskID, selection.profileID) { + cause := fmt.Errorf("profile #%d failed %d consecutive calls (last: %w)", selection.profileID, llmFailStreakThreshold, callErr) + transition, markErr := hooks.exhaust(selection, cause) + if markErr != nil { + return llm.Message{}, "", llm.Usage{}, fmt.Errorf("mark profile exhausted after %v: %w", callErr, markErr) + } + if transition.Advanced && !transition.Stale && hooks.transition != nil { + hooks.transition(selection, transition, callErr) + } + if !transition.Stale && transition.NextProfileID == nil { + return llm.Message{}, "", llm.Usage{}, &taskLLMError{taskID: taskID, chainExhausted: transition.ChainExhausted, cause: callErr} + } + continue // replay the same logical request on the next profile + } + return llm.Message{}, "", llm.Usage{}, callErr + } // profileID=0: the explicit chain was cleared; non-quota error: pass through. Neither changes the task failover state. if selection.profileID == 0 || !isQuotaExhaustedError(callErr) { return llm.Message{}, "", llm.Usage{}, callErr @@ -338,6 +399,7 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest break } if streamErr == nil { + resetLLMFailStreak(taskID, selection.profileID) for _, buffered := range pending { if !yield(buffered, nil) { return @@ -345,6 +407,28 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest } return } + // Non-quota failures on a chain profile escalate after a consecutive + // streak (#99), mirroring completeTaskLLM. Escalation only applies to + // uncommitted streams: once events reached the caller, replaying on a + // new profile could duplicate delivered output, so the error surfaces. + if selection.profileID != 0 && !committed && !isQuotaExhaustedError(streamErr) && ctx.Err() == nil { + if noteLLMCallFailure(taskID, selection.profileID) { + cause := fmt.Errorf("profile #%d failed %d consecutive calls (last: %w)", selection.profileID, llmFailStreakThreshold, streamErr) + transition, markErr := hooks.exhaust(selection, cause) + if markErr != nil { + yield(llm.StreamEvent{}, fmt.Errorf("mark profile exhausted after %v: %w", streamErr, markErr)) + return + } + if transition.Advanced && !transition.Stale && hooks.transition != nil { + hooks.transition(selection, transition, streamErr) + } + if !transition.Stale && transition.NextProfileID == nil { + yield(llm.StreamEvent{}, &taskLLMError{taskID: taskID, chainExhausted: transition.ChainExhausted, cause: streamErr}) + return + } + continue // nothing delivered — safe to replay on the next profile + } + } // profileID=0 means the explicit chain was cleared while this stable task // bundle was still in use. Agent/global fallback errors follow the legacy // behavior and never mutate task failover state. diff --git a/server/task_llm_escalation_test.go b/server/task_llm_escalation_test.go new file mode 100644 index 0000000..725b24d --- /dev/null +++ b/server/task_llm_escalation_test.go @@ -0,0 +1,85 @@ +package server + +import ( + "context" + "errors" + "iter" + "testing" + + "github.com/Autumn-27/artex/db" + "github.com/Autumn-27/norma/llm" +) + +type failingProvider struct{ err error } + +func (f *failingProvider) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { + return nil +} + +func (f *failingProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { + return llm.Message{}, "", llm.Usage{}, f.err +} + +func TestEscalationAdvancesAfterConsecutiveFailures(t *testing.T) { + resetAllLLMFailStreaksForTest() + prov := &failingProvider{err: errors.New("boom: upstream 500")} + profile := int64(7) + nextProfile := int64(8) + exhaustCalls := 0 + hooks := taskLLMStreamHooks{ + current: func() (taskLLMSelection, error) { + return taskLLMSelection{task: &Task{ID: "1"}, profileID: profile, revision: 1, provider: prov}, nil + }, + exhaust: func(sel taskLLMSelection, cause error) (db.TaskLLMTransition, error) { + exhaustCalls++ + if exhaustCalls == 1 { + return db.TaskLLMTransition{Advanced: true, NextProfileID: &nextProfile}, nil + } + return db.TaskLLMTransition{ChainExhausted: true}, nil + }, + } + // first two failures: passthrough, chain untouched + for i := 1; i <= 2; i++ { + _, _, _, err := completeTaskLLM(context.Background(), "1", llm.CompletionRequest{}, hooks) + if err == nil || errors.Is(err, err) == false { + t.Fatalf("call %d: expected the raw failure", i) + } + if exhaustCalls != 0 { + t.Fatalf("call %d: exhaust fired early (%d)", i, exhaustCalls) + } + } + // third consecutive failure: escalates, advances to profile 8, which itself + // fails (streak 1) and returns the raw error + _, _, _, err := completeTaskLLM(context.Background(), "1", llm.CompletionRequest{}, hooks) + if err == nil { + t.Fatal("expected the follow-up failure from the next profile") + } + if exhaustCalls != 1 { + t.Fatalf("exhaust calls=%d want 1", exhaustCalls) + } + // The escalation inside call 3 already replayed once on the same profile + // (streak 1); the fake current() keeps reporting profile 7, so the streak + // key is unchanged. One more failure reaches streak 2, and the NEXT one + // fires the second escalation, which the fake exhaust answers with a fully + // exhausted chain. + _, _, _, err = completeTaskLLM(context.Background(), "1", llm.CompletionRequest{}, hooks) + if err == nil { + t.Fatal("expected raw failure at streak 2") + } + if exhaustCalls != 1 { + t.Fatalf("exhaust calls=%d want 1 at streak 2", exhaustCalls) + } + _, _, _, err = completeTaskLLM(context.Background(), "1", llm.CompletionRequest{}, hooks) + if !isTaskLLMChainExhausted(err) { + t.Fatalf("expected chain-exhausted classification, got %v", err) + } + if exhaustCalls != 2 { + t.Fatalf("exhaust calls=%d want 2", exhaustCalls) + } +} + +func resetAllLLMFailStreaksForTest() { + llmFailStreaks.Lock() + llmFailStreaks.m = map[string]int{} + llmFailStreaks.Unlock() +}