mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-13 11:40:53 +08:00
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
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user