mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-15 04:30:55 +08:00
Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b48c58d3b6 | ||
|
|
4dcba3d07e | ||
|
|
a195dfee2a | ||
|
|
ff017216bf | ||
|
|
1ec2511c4b | ||
|
|
b757fc7e62 | ||
|
|
72b79af758 | ||
|
|
579012e09d | ||
|
|
6894c391dd | ||
|
|
89348b5512 | ||
|
|
2fd6ed08fc | ||
|
|
b237fbaa32 | ||
|
|
d24e734f19 |
+34
@@ -21,6 +21,40 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
|
|||||||
&& curl -fsSL https://deb.nodesource.com/setup_20.x | bash - \
|
&& curl -fsSL https://deb.nodesource.com/setup_20.x | bash - \
|
||||||
&& apt-get install -y --no-install-recommends nodejs \
|
&& apt-get install -y --no-install-recommends nodejs \
|
||||||
&& rm -rf /var/lib/apt/lists/*
|
&& rm -rf /var/lib/apt/lists/*
|
||||||
|
# Web-recon toolbox (drost-community take 1, #100): pinned release binaries installed
|
||||||
|
# into /usr/local/bin so hunt workers can run them directly through their Bash tool.
|
||||||
|
# Versions pinned per tool; upgrade deliberately. sqlmap is a python checkout (no
|
||||||
|
# release binaries) symlinked into PATH.
|
||||||
|
ARG NUCLEI_VERSION=v3.11.1
|
||||||
|
ARG HTTPX_VERSION=v1.12.0
|
||||||
|
ARG SUBFINDER_VERSION=v2.17.0
|
||||||
|
ARG KATANA_VERSION=v1.8.0
|
||||||
|
ARG FFUF_VERSION=v2.3.0
|
||||||
|
ARG DALFOX_VERSION=v3.2.4
|
||||||
|
RUN set -eux; \
|
||||||
|
base="https://github.com"; \
|
||||||
|
pd_fetch() { repo="$1"; ver="$2"; name="$3"; nov="${ver#v}"; \
|
||||||
|
curl -fsSL "$base/$repo/releases/download/$ver/${name}_${nov}_linux_${TARGETARCH}.zip" -o /tmp/t.zip \
|
||||||
|
&& unzip -o -q /tmp/t.zip -d /tmp/td \
|
||||||
|
&& find /tmp/td -type f -name "$name" -exec install -m 0755 {} /usr/local/bin/ \; \
|
||||||
|
&& rm -rf /tmp/t.zip /tmp/td; }; \
|
||||||
|
case "$TARGETARCH" in amd64) FFUF_ARCH=amd64; DALFOX_ARCH=x86_64;; arm64) FFUF_ARCH=arm64; DALFOX_ARCH=aarch64;; esac; \
|
||||||
|
pd_fetch projectdiscovery/nuclei "$NUCLEI_VERSION" nuclei; \
|
||||||
|
pd_fetch projectdiscovery/httpx "$HTTPX_VERSION" httpx; \
|
||||||
|
pd_fetch projectdiscovery/subfinder "$SUBFINDER_VERSION" subfinder; \
|
||||||
|
pd_fetch projectdiscovery/katana "$KATANA_VERSION" katana; \
|
||||||
|
tar_fetch() { url="$1"; name="$2"; \
|
||||||
|
curl -fsSL "$url" -o /tmp/t.tgz \
|
||||||
|
&& mkdir -p /tmp/td && tar -xzf /tmp/t.tgz -C /tmp/td \
|
||||||
|
&& find /tmp/td -type f -name "$name" -exec install -m 0755 {} /usr/local/bin/ \; \
|
||||||
|
&& rm -rf /tmp/t.tgz /tmp/td; }; \
|
||||||
|
tar_fetch "$base/ffuf/ffuf/releases/download/$FFUF_VERSION/ffuf_${FFUF_VERSION#v}_linux_${FFUF_ARCH}.tar.gz" ffuf; \
|
||||||
|
tar_fetch "$base/hahwul/dalfox/releases/download/$DALFOX_VERSION/dalfox-${DALFOX_VERSION}-linux-${DALFOX_ARCH}-musl.tar.gz" dalfox; \
|
||||||
|
git clone --depth 1 https://github.com/sqlmapproject/sqlmap /opt/sqlmap; \
|
||||||
|
ln -s /opt/sqlmap/sqlmap.py /usr/local/bin/sqlmap; \
|
||||||
|
nuclei -version >/dev/null; httpx -version; subfinder -version; katana -version; \
|
||||||
|
ffuf -V >/dev/null; dalfox --version >/dev/null 2>&1 || true; sqlmap --version >/dev/null
|
||||||
|
|
||||||
# Preinstall Playwright MCP and CLI (globally) so no npx network download is needed at runtime.
|
# Preinstall Playwright MCP and CLI (globally) so no npx network download is needed at runtime.
|
||||||
# @playwright/mcp: browser MCP runs directly via `npx @playwright/mcp` (installed globally, no -y/@latest needed).
|
# @playwright/mcp: browser MCP runs directly via `npx @playwright/mcp` (installed globally, no -y/@latest needed).
|
||||||
# @playwright/cli: provides playwright-cli, verified with --help right after install.
|
# @playwright/cli: provides playwright-cli, verified with --help right after install.
|
||||||
|
|||||||
@@ -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/
|
// upstream nodes it derived from: facts/intents/findings) and yields (the facts/
|
||||||
// findings it produced) — without pulling full payloads. parentsOf/yieldsOf are
|
// findings it produced) — without pulling full payloads. parentsOf/yieldsOf are
|
||||||
// built from the exploration edges in graph_overview.
|
// 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 {
|
func compactIntents(ns []*db.Node, parentsOf, yieldsOf map[int64][]int64) []map[string]any {
|
||||||
out := make([]map[string]any, 0, len(ns))
|
out := make([]map[string]any, 0, len(ns))
|
||||||
for _, n := range 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
|
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["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)
|
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
|
// 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
|
// 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
|
// the recently confirmed findings at a glance each round when judging goals; use
|
||||||
|
|||||||
@@ -522,6 +522,7 @@ CREATE TABLE IF NOT EXISTS tasks (
|
|||||||
llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL,
|
llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL,
|
||||||
active_llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL,
|
active_llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL,
|
||||||
llm_chain_revision BIGINT NOT NULL DEFAULT 0,
|
llm_chain_revision BIGINT NOT NULL DEFAULT 0,
|
||||||
|
llm_rotate BOOLEAN NOT NULL DEFAULT false,
|
||||||
company_id BIGINT REFERENCES companies(id) ON DELETE SET NULL,
|
company_id BIGINT REFERENCES companies(id) ON DELETE SET NULL,
|
||||||
parent_ref TEXT,
|
parent_ref TEXT,
|
||||||
timeout_seconds INTEGER NOT NULL DEFAULT 0,
|
timeout_seconds INTEGER NOT NULL DEFAULT 0,
|
||||||
@@ -545,6 +546,7 @@ CREATE TRIGGER trg_tasks_upd BEFORE UPDATE ON tasks
|
|||||||
-- planner heartbeat trigger interval (seconds); backfills old databases. Default 300s (5min). See docs/planner-trigger-impl-plan.md
|
-- planner heartbeat trigger interval (seconds); backfills old databases. Default 300s (5min). See docs/planner-trigger-impl-plan.md
|
||||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300;
|
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300;
|
||||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_quiet_mode BOOLEAN NOT NULL DEFAULT true;
|
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_quiet_mode BOOLEAN NOT NULL DEFAULT true;
|
||||||
|
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS llm_rotate BOOLEAN NOT NULL DEFAULT false;
|
||||||
-- Concurrency-limit hold state; backfills old databases. true=queued due to the concurrency limit, auto-starts when a slot frees up.
|
-- Concurrency-limit hold state; backfills old databases. true=queued due to the concurrency limit, auto-starts when a slot frees up.
|
||||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS queued BOOLEAN NOT NULL DEFAULT false;
|
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS queued BOOLEAN NOT NULL DEFAULT false;
|
||||||
-- Asset coverage feature switch; backfills old databases. true (default)=compute/show test coverage, auto-accumulate test scope,
|
-- Asset coverage feature switch; backfills old databases. true (default)=compute/show test coverage, auto-accumulate test scope,
|
||||||
|
|||||||
+17
-5
@@ -33,6 +33,9 @@ type Task struct {
|
|||||||
LLMChainRevision int64 `json:"-"`
|
LLMChainRevision int64 `json:"-"`
|
||||||
LLMFailoverState string `json:"llm_failover_state,omitempty"`
|
LLMFailoverState string `json:"llm_failover_state,omitempty"`
|
||||||
LLMFailoverReason string `json:"llm_failover_reason,omitempty"`
|
LLMFailoverReason string `json:"llm_failover_reason,omitempty"`
|
||||||
|
// LLMRotate: per-intent chain-head rotation switch (#85); false = classic
|
||||||
|
// ordered failover from the active head.
|
||||||
|
LLMRotate bool `json:"llm_rotate"`
|
||||||
SourceTaskIDs []int64 `json:"source_task_ids,omitempty"`
|
SourceTaskIDs []int64 `json:"source_task_ids,omitempty"`
|
||||||
CompanyIDs []int64 `json:"company_ids,omitempty"`
|
CompanyIDs []int64 `json:"company_ids,omitempty"`
|
||||||
ParentRef string `json:"parent_ref,omitempty"` // parent task id (recorded by orchestration spawn; empty=top level)
|
ParentRef string `json:"parent_ref,omitempty"` // parent task id (recorded by orchestration spawn; empty=top level)
|
||||||
@@ -165,6 +168,11 @@ type TaskCreateOptions struct {
|
|||||||
CoverageEnabled *bool
|
CoverageEnabled *bool
|
||||||
// InterceptRules are task-level asset interception rules, written into task_intercept_rules in the same transaction as the task at creation.
|
// InterceptRules are task-level asset interception rules, written into task_intercept_rules in the same transaction as the task at creation.
|
||||||
InterceptRules []TaskInterceptRuleInput
|
InterceptRules []TaskInterceptRuleInput
|
||||||
|
// LLMRotate enables per-intent round-robin rotation of the task's LLM chain
|
||||||
|
// heads (#85): each claimed worker intent runs on the next profile in the
|
||||||
|
// chain instead of the single active head. nil = off (default), preserving
|
||||||
|
// the ordered-failover semantics.
|
||||||
|
LLMRotate *bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// CreateTaskWithOptions creates an exploration, task, direct source relations,
|
// CreateTaskWithOptions creates an exploration, task, direct source relations,
|
||||||
@@ -239,10 +247,14 @@ VALUES ($1, 'fact', $2, 0, 'origin', 'system')`, expID, string(originPayload));
|
|||||||
PlanQuietMode: quietMode,
|
PlanQuietMode: quietMode,
|
||||||
CoverageEnabled: coverageEnabled,
|
CoverageEnabled: coverageEnabled,
|
||||||
}
|
}
|
||||||
|
rotate := false
|
||||||
|
if opts.LLMRotate != nil {
|
||||||
|
rotate = *opts.LLMRotate
|
||||||
|
}
|
||||||
if err := tx.QueryRow(`
|
if err := tx.QueryRow(`
|
||||||
INSERT INTO tasks(name, category_id, description, goal, exploration_id, llm_profile_id, active_llm_profile_id, timeout_seconds, plan_heartbeat_seconds, plan_quiet_mode, coverage_enabled)
|
INSERT INTO tasks(name, category_id, description, goal, exploration_id, llm_profile_id, active_llm_profile_id, timeout_seconds, plan_heartbeat_seconds, plan_quiet_mode, coverage_enabled, llm_rotate)
|
||||||
VALUES ($1,$2,$3,$4,$5,$6,$6,$7,$8,$9,$10)
|
VALUES ($1,$2,$3,$4,$5,$6,$6,$7,$8,$9,$10,$11)
|
||||||
RETURNING id, status, paused, created_at`, opts.Name, opts.CategoryID, description, goal, expID, active, opts.TimeoutSeconds, opts.PlanHeartbeatSeconds, quietMode, coverageEnabled).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil {
|
RETURNING id, status, paused, created_at`, opts.Name, opts.CategoryID, description, goal, expID, active, opts.TimeoutSeconds, opts.PlanHeartbeatSeconds, quietMode, coverageEnabled, rotate).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if err := insertTaskRelations(tx, t.ID, opts.SourceTaskIDs); err != nil {
|
if err := insertTaskRelations(tx, t.ID, opts.SourceTaskIDs); err != nil {
|
||||||
@@ -358,11 +370,11 @@ func insertTaskLLMProfiles(tx *sql.Tx, taskID int64, profileIDs []int64) error {
|
|||||||
|
|
||||||
const taskCols = `id, COALESCE(name,''), category_id,
|
const taskCols = `id, COALESCE(name,''), category_id,
|
||||||
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
|
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
|
||||||
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(plan_quiet_mode,true), COALESCE(coverage_enabled,true), first_run_at, deadline_at`
|
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(plan_quiet_mode,true), COALESCE(coverage_enabled,true), COALESCE(llm_rotate,false), first_run_at, deadline_at`
|
||||||
|
|
||||||
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
|
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
|
||||||
var t Task
|
var t Task
|
||||||
if err := sc.Scan(&t.ID, &t.Name, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.PlanQuietMode, &t.CoverageEnabled, &t.FirstRunAt, &t.DeadlineAt); err != nil {
|
if err := sc.Scan(&t.ID, &t.Name, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.PlanQuietMode, &t.CoverageEnabled, &t.LLMRotate, &t.FirstRunAt, &t.DeadlineAt); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
t.Pinned = t.PinnedAt != nil
|
t.Pinned = t.PinnedAt != nil
|
||||||
|
|||||||
+1
-1
@@ -62,7 +62,7 @@ func TestCoreTaskLifecyclePG(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// create a task → 201, returns PG task (string id + exploration_id)
|
// create a task → 201, returns PG task (string id + exploration_id)
|
||||||
code, out := doRetry("POST", "/api/tasks", map[string]string{"description": "smoke", "goal": "test SQLi/XSS"})
|
code, out := doRetry("POST", "/api/tasks", map[string]string{"description": "smoke", "goal": "Probe the lab target at http://127.0.0.1:9/v1 for unauthenticated data exposure; record one record as evidence and stop."})
|
||||||
if code != 201 {
|
if code != 201 {
|
||||||
t.Fatalf("create task: %d (%v)", code, out)
|
t.Fatalf("create task: %d (%v)", code, out)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -55,6 +55,7 @@ type TaskDTO struct {
|
|||||||
LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // LLM profile used for this task; nil = default
|
LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // LLM profile used for this task; nil = default
|
||||||
LLMProfileIDs []int64 `json:"llm_profile_ids"`
|
LLMProfileIDs []int64 `json:"llm_profile_ids"`
|
||||||
ActiveLLMProfileID *int64 `json:"active_llm_profile_id,omitempty"`
|
ActiveLLMProfileID *int64 `json:"active_llm_profile_id,omitempty"`
|
||||||
|
LLMRotate bool `json:"llm_rotate"`
|
||||||
LLMFailoverState string `json:"llm_failover_state"`
|
LLMFailoverState string `json:"llm_failover_state"`
|
||||||
LLMFailoverReason string `json:"llm_failover_reason,omitempty"`
|
LLMFailoverReason string `json:"llm_failover_reason,omitempty"`
|
||||||
SourceTaskIDs []string `json:"source_task_ids"`
|
SourceTaskIDs []string `json:"source_task_ids"`
|
||||||
@@ -130,6 +131,7 @@ func taskDTO(t *Task, status string) TaskDTO {
|
|||||||
LLMProfileID: llmState.ProfileID,
|
LLMProfileID: llmState.ProfileID,
|
||||||
LLMProfileIDs: profileIDs,
|
LLMProfileIDs: profileIDs,
|
||||||
ActiveLLMProfileID: llmState.ActiveID,
|
ActiveLLMProfileID: llmState.ActiveID,
|
||||||
|
LLMRotate: t.LLMRotate,
|
||||||
LLMFailoverState: llmState.FailoverState,
|
LLMFailoverState: llmState.FailoverState,
|
||||||
LLMFailoverReason: llmState.FailoverReason,
|
LLMFailoverReason: llmState.FailoverReason,
|
||||||
SourceTaskIDs: sourceIDs,
|
SourceTaskIDs: sourceIDs,
|
||||||
|
|||||||
@@ -1017,6 +1017,14 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker
|
|||||||
// must already hold one task-operation admission for the whole sequence so a delete
|
// must already hold one task-operation admission for the whole sequence so a delete
|
||||||
// cannot observe quiescence between the LLM return and the final DB writes.
|
// cannot observe quiescence between the LLM return and the final DB writes.
|
||||||
func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *agent.Worker, intent *db.Node, requestID, message string) bool {
|
func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *agent.Worker, intent *db.Node, requestID, message string) bool {
|
||||||
|
// Per-intent LLM rotation (#85): each claimed worker intent runs on the
|
||||||
|
// next chain head. The pin travels through the run context into the
|
||||||
|
// provider resolver; non-worker roles and failover keep the active head.
|
||||||
|
if t.LLMRotate {
|
||||||
|
if head, ok := t.nextRotatedProfile(); ok {
|
||||||
|
ctx = withLLMProfilePin(ctx, head)
|
||||||
|
}
|
||||||
|
}
|
||||||
hasChatMessage := message != ""
|
hasChatMessage := message != ""
|
||||||
e.stampFirstRun(t) // first real execution → stamp first_run_at + compute the deadline (only for tasks with a timeout)
|
e.stampFirstRun(t) // first real execution → stamp first_run_at + compute the deadline (only for tasks with a timeout)
|
||||||
e.touch(t.ID)
|
e.touch(t.ID)
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+21
-1
@@ -51,6 +51,10 @@ type Task struct {
|
|||||||
ActiveLLMProfileID *int64 `json:"active_llm_profile_id,omitempty"`
|
ActiveLLMProfileID *int64 `json:"active_llm_profile_id,omitempty"`
|
||||||
LLMChainRevision int64 `json:"-"`
|
LLMChainRevision int64 `json:"-"`
|
||||||
LLMFailoverState string `json:"llm_failover_state,omitempty"`
|
LLMFailoverState string `json:"llm_failover_state,omitempty"`
|
||||||
|
// LLMRotate (#85): per-intent round-robin over the task's LLM chain heads.
|
||||||
|
LLMRotate bool `json:"llm_rotate,omitempty"`
|
||||||
|
rotateMu sync.Mutex
|
||||||
|
rotateNext int
|
||||||
LLMFailoverReason string `json:"llm_failover_reason,omitempty"`
|
LLMFailoverReason string `json:"llm_failover_reason,omitempty"`
|
||||||
SourceTaskIDs []int64 `json:"source_task_ids,omitempty"`
|
SourceTaskIDs []int64 `json:"source_task_ids,omitempty"`
|
||||||
CompanyIDs []int64 `json:"company_ids,omitempty"`
|
CompanyIDs []int64 `json:"company_ids,omitempty"`
|
||||||
@@ -178,6 +182,22 @@ func (t *Task) llmStateSnapshot() taskLLMState {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// nextRotatedProfile returns the chain head for the NEXT claimed worker
|
||||||
|
// intent under per-intent rotation (#85): strict round-robin over the task's
|
||||||
|
// chain. The active head stays authoritative for every non-worker role and for
|
||||||
|
// failover ordering; rotation only distributes worker lanes across heads.
|
||||||
|
func (t *Task) nextRotatedProfile() (int64, bool) {
|
||||||
|
t.rotateMu.Lock()
|
||||||
|
defer t.rotateMu.Unlock()
|
||||||
|
ids := t.LLMProfileIDs
|
||||||
|
if len(ids) == 0 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
pick := ids[t.rotateNext%len(ids)]
|
||||||
|
t.rotateNext++
|
||||||
|
return pick, true
|
||||||
|
}
|
||||||
|
|
||||||
func (t *Task) setLLMState(profileID, activeID *int64, profileIDs []int64, revision int64, state, reason string) bool {
|
func (t *Task) setLLMState(profileID, activeID *int64, profileIDs []int64, revision int64, state, reason string) bool {
|
||||||
t.llmMu.Lock()
|
t.llmMu.Lock()
|
||||||
defer t.llmMu.Unlock()
|
defer t.llmMu.Unlock()
|
||||||
@@ -840,7 +860,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
|
|||||||
QueuedAt: unixNanoOrZero(pt.QueuedAt), QueueMode: pt.QueueMode,
|
QueuedAt: unixNanoOrZero(pt.QueuedAt), QueueMode: pt.QueueMode,
|
||||||
CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef,
|
CompletedAt: unixOrZero(pt.CompletedAt), Status: pt.Status, ParentRef: pt.ParentRef,
|
||||||
LLMProfileID: pt.LLMProfileID,
|
LLMProfileID: pt.LLMProfileID,
|
||||||
LLMProfileIDs: append([]int64(nil), pt.LLMProfileIDs...), ActiveLLMProfileID: pt.ActiveLLMProfileID,
|
LLMProfileIDs: append([]int64(nil), pt.LLMProfileIDs...), ActiveLLMProfileID: pt.ActiveLLMProfileID, LLMRotate: pt.LLMRotate,
|
||||||
LLMChainRevision: pt.LLMChainRevision,
|
LLMChainRevision: pt.LLMChainRevision,
|
||||||
LLMFailoverState: pt.LLMFailoverState, LLMFailoverReason: pt.LLMFailoverReason,
|
LLMFailoverState: pt.LLMFailoverState, LLMFailoverReason: pt.LLMFailoverReason,
|
||||||
SourceTaskIDs: append([]int64(nil), pt.SourceTaskIDs...),
|
SourceTaskIDs: append([]int64(nil), pt.SourceTaskIDs...),
|
||||||
|
|||||||
+7
-1
@@ -1506,7 +1506,8 @@ type createTaskReq struct {
|
|||||||
LLMProfileIDs []int64 `json:"llm_profile_ids,omitempty"` // ordered task-level profile chain; the first entry takes effect initially
|
LLMProfileIDs []int64 `json:"llm_profile_ids,omitempty"` // ordered task-level profile chain; the first entry takes effect initially
|
||||||
SourceTaskIDs []string `json:"source_task_ids,omitempty"` // direct, read-only inherited source tasks
|
SourceTaskIDs []string `json:"source_task_ids,omitempty"` // direct, read-only inherited source tasks
|
||||||
CompanyIDs []int64 `json:"company_ids,omitempty"` // links the company scope and snapshots its current assets; does not copy assets or force intent generation
|
CompanyIDs []int64 `json:"company_ids,omitempty"` // links the company scope and snapshots its current assets; does not copy assets or force intent generation
|
||||||
TimeoutSeconds int `json:"timeout_seconds"` // task-level timeout (seconds); 0/omitted = no limit
|
TimeoutSeconds int `json:"timeout_seconds"`
|
||||||
|
LLMRotate *bool `json:"llm_rotate"` // task-level timeout (seconds); 0/omitted = no limit
|
||||||
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner heartbeat interval (seconds); 0/omitted = default 600 (10min); floor = default = 600, lower values are raised to 600
|
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner heartbeat interval (seconds); 0/omitted = default 600 (10min); floor = default = 600, lower values are raised to 600
|
||||||
PlanQuietMode *bool `json:"plan_quiet_mode,omitempty"` // planner quiet-mode switch (#86); omitted = default on; false = edge rounds always plan
|
PlanQuietMode *bool `json:"plan_quiet_mode,omitempty"` // planner quiet-mode switch (#86); omitted = default on; false = edge rounds always plan
|
||||||
SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // issue one seed intent at creation (content = description + goal) so a worker can start right away without waiting for the first planner round; omitted/null = default off, the standard plan-then-execute flow. Pass true explicitly to enable (for CTF tasks often solved in one work, this skips the pre-start planner round).
|
SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // issue one seed intent at creation (content = description + goal) so a worker can start right away without waiting for the first planner round; omitted/null = default off, the standard plan-then-execute flow. Pass true explicitly to enable (for CTF tasks often solved in one work, this skips the pre-start planner round).
|
||||||
@@ -1541,6 +1542,10 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) {
|
|||||||
writeErr(w, 400, err.Error())
|
writeErr(w, 400, err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if err := goalGateCheck(req.Goal); err != nil {
|
||||||
|
writeErr(w, 400, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
if req.TimeoutSeconds < 0 {
|
if req.TimeoutSeconds < 0 {
|
||||||
req.TimeoutSeconds = 0
|
req.TimeoutSeconds = 0
|
||||||
}
|
}
|
||||||
@@ -1580,6 +1585,7 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) {
|
|||||||
TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds, PlanQuietMode: req.PlanQuietMode,
|
TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds, PlanQuietMode: req.PlanQuietMode,
|
||||||
CoverageEnabled: req.CoverageEnabled,
|
CoverageEnabled: req.CoverageEnabled,
|
||||||
InterceptRules: interceptRules,
|
InterceptRules: interceptRules,
|
||||||
|
LLMRotate: req.LLMRotate,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, db.ErrTaskCategoryInvalid) || errors.Is(err, db.ErrTaskCategoryNotFound) {
|
if errors.Is(err, db.ErrTaskCategoryInvalid) || errors.Is(err, db.ErrTaskCategoryNotFound) {
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ func TestTaskCredentialsFlow(t *testing.T) {
|
|||||||
}
|
}
|
||||||
resp := request("POST", "/api/tasks", `{
|
resp := request("POST", "/api/tasks", `{
|
||||||
"description": "credential flow test",
|
"description": "credential flow test",
|
||||||
"goal": "verify task credentials",
|
"goal": "Verify credential handling against the lab target at http://127.0.0.1:9/v1 and record the outcome.",
|
||||||
"credentials": [
|
"credentials": [
|
||||||
{"domain": "target.example.com", "username": "alias@test", "password": "s3cret!", "note": "program alias account"},
|
{"domain": "target.example.com", "username": "alias@test", "password": "s3cret!", "note": "program alias account"},
|
||||||
{"domain": "target.example.com", "username": "", "password": ""}
|
{"domain": "target.example.com", "username": "", "password": ""}
|
||||||
@@ -32,7 +32,7 @@ func TestTaskCredentialsFlow(t *testing.T) {
|
|||||||
|
|
||||||
resp = request("POST", "/api/tasks", `{
|
resp = request("POST", "/api/tasks", `{
|
||||||
"description": "credential flow test",
|
"description": "credential flow test",
|
||||||
"goal": "verify task credentials",
|
"goal": "Verify credential handling against the lab target at http://127.0.0.1:9/v1 and record the outcome.",
|
||||||
"credentials": [
|
"credentials": [
|
||||||
{"domain": "target.example.com", "username": "alias@test", "password": "s3cret!", "note": "program alias account"},
|
{"domain": "target.example.com", "username": "alias@test", "password": "s3cret!", "note": "program alias account"},
|
||||||
{"domain": "app.target.example.com", "username": "agent-account", "password": ""}
|
{"domain": "app.target.example.com", "username": "agent-account", "password": ""}
|
||||||
|
|||||||
+129
-4
@@ -10,6 +10,7 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/Autumn-27/artex/agent"
|
"github.com/Autumn-27/artex/agent"
|
||||||
@@ -85,6 +86,54 @@ func isQuotaExhaustedError(err error) bool {
|
|||||||
return err != nil && agent.IsQuotaExhaustedMessage(err.Error())
|
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
|
||||||
|
}
|
||||||
|
|
||||||
|
// resetLLMFailStreaksForTest clears every streak; test isolation only (the
|
||||||
|
// escalation counter is package-global in-memory state).
|
||||||
|
func resetLLMFailStreaksForTest() {
|
||||||
|
llmFailStreaks.Lock()
|
||||||
|
llmFailStreaks.m = map[string]int{}
|
||||||
|
llmFailStreaks.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
func resetLLMFailStreak(taskID string, profileID int64) {
|
||||||
|
llmFailStreaks.Lock()
|
||||||
|
defer llmFailStreaks.Unlock()
|
||||||
|
delete(llmFailStreaks.m, llmFailKey(taskID, profileID))
|
||||||
|
}
|
||||||
|
|
||||||
type taskLLMSelection struct {
|
type taskLLMSelection struct {
|
||||||
task *Task
|
task *Task
|
||||||
profileID int64
|
profileID int64
|
||||||
@@ -95,8 +144,25 @@ type taskLLMSelection struct {
|
|||||||
retry agent.RetryConfig
|
retry agent.RetryConfig
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// llmProfilePin carries a per-run chain-head override through the call stack
|
||||||
|
// (#85 per-intent rotation). The engine sets it when launching a worker run;
|
||||||
|
// resolution stays within the task's chain and never bypasses failover state.
|
||||||
|
type llmProfilePinKey struct{}
|
||||||
|
|
||||||
|
func withLLMProfilePin(ctx context.Context, profileID int64) context.Context {
|
||||||
|
return context.WithValue(ctx, llmProfilePinKey{}, profileID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func llmProfilePinFrom(ctx context.Context) (int64, bool) {
|
||||||
|
if ctx == nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
id, ok := ctx.Value(llmProfilePinKey{}).(int64)
|
||||||
|
return id, ok
|
||||||
|
}
|
||||||
|
|
||||||
type taskLLMStreamHooks struct {
|
type taskLLMStreamHooks struct {
|
||||||
current func() (taskLLMSelection, error)
|
current func(context.Context) (taskLLMSelection, error)
|
||||||
exhaust func(taskLLMSelection, error) (db.TaskLLMTransition, error)
|
exhaust func(taskLLMSelection, error) (db.TaskLLMTransition, error)
|
||||||
transition func(taskLLMSelection, db.TaskLLMTransition, error)
|
transition func(taskLLMSelection, db.TaskLLMTransition, error)
|
||||||
}
|
}
|
||||||
@@ -108,6 +174,14 @@ type taskLLMStreamHooks struct {
|
|||||||
// to the task chain, and to the global configuration when the chain is empty.
|
// to the task chain, and to the global configuration when the chain is empty.
|
||||||
// The returned profile id is non-zero only on the task-chain path — streamTaskLLM uses it to decide whether a quota error should advance the task's failover state (binding/global paths leave the chain state untouched, preserving existing semantics).
|
// The returned profile id is non-zero only on the task-chain path — streamTaskLLM uses it to decide whether a quota error should advance the task's failover state (binding/global paths leave the chain state untouched, preserving existing semantics).
|
||||||
func (r *taskLLMRuntime) current() (taskLLMSelection, error) {
|
func (r *taskLLMRuntime) current() (taskLLMSelection, error) {
|
||||||
|
return r.currentFor(context.Background())
|
||||||
|
}
|
||||||
|
|
||||||
|
// currentFor resolves like current, honoring a per-run profile pin set by the
|
||||||
|
// engine for per-intent rotation (#85). A pin is only a dispatch preference:
|
||||||
|
// the chain must still be in a ready state (an exhausted chain with no active
|
||||||
|
// head stays exhausted) and the pinned id must belong to the task's chain.
|
||||||
|
func (r *taskLLMRuntime) currentFor(ctx context.Context) (taskLLMSelection, error) {
|
||||||
taskNum, err := parseTaskID(r.taskID)
|
taskNum, err := parseTaskID(r.taskID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return taskLLMSelection{}, err
|
return taskLLMSelection{}, err
|
||||||
@@ -131,6 +205,14 @@ func (r *taskLLMRuntime) current() (taskLLMSelection, error) {
|
|||||||
return sel, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")}
|
return sel, &taskLLMError{taskID: r.taskID, chainExhausted: true, cause: errors.New("all selected profiles are quota exhausted")}
|
||||||
}
|
}
|
||||||
sel.profileID = *pt.ActiveLLMProfileID
|
sel.profileID = *pt.ActiveLLMProfileID
|
||||||
|
if pin, ok := llmProfilePinFrom(ctx); ok && pin != sel.profileID {
|
||||||
|
for _, id := range pt.LLMProfileIDs {
|
||||||
|
if id == pin {
|
||||||
|
sel.profileID = pin
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
prov, cfg, ok := r.s.providerForProfile(sel.profileID)
|
prov, cfg, ok := r.s.providerForProfile(sel.profileID)
|
||||||
if !ok {
|
if !ok {
|
||||||
return sel, fmt.Errorf("LLM profile #%d is missing or invalid", sel.profileID)
|
return sel, fmt.Errorf("LLM profile #%d is missing or invalid", sel.profileID)
|
||||||
@@ -201,7 +283,7 @@ func parseTaskID(id string) (int64, error) {
|
|||||||
// exhausted, and how to emit a failover transition.
|
// exhausted, and how to emit a failover transition.
|
||||||
func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks {
|
func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks {
|
||||||
return taskLLMStreamHooks{
|
return taskLLMStreamHooks{
|
||||||
current: r.current,
|
current: func(ctx context.Context) (taskLLMSelection, error) { return r.currentFor(ctx) },
|
||||||
exhaust: func(selection taskLLMSelection, cause error) (db.TaskLLMTransition, error) {
|
exhaust: func(selection taskLLMSelection, cause error) (db.TaskLLMTransition, error) {
|
||||||
taskNum, _ := parseTaskID(r.taskID)
|
taskNum, _ := parseTaskID(r.taskID)
|
||||||
transition, err := r.s.m.pg.MarkTaskLLMProfileQuotaExhaustedAtRevision(taskNum, selection.profileID, selection.revision, cause.Error())
|
transition, err := r.s.m.pg.MarkTaskLLMProfileQuotaExhaustedAtRevision(taskNum, selection.profileID, selection.revision, cause.Error())
|
||||||
@@ -236,7 +318,7 @@ func (r *taskLLMRuntime) Complete(ctx context.Context, req llm.CompletionRequest
|
|||||||
|
|
||||||
func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) (llm.Message, string, llm.Usage, error) {
|
func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) (llm.Message, string, llm.Usage, error) {
|
||||||
for {
|
for {
|
||||||
selection, err := hooks.current()
|
selection, err := hooks.current(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return llm.Message{}, "", llm.Usage{}, err
|
return llm.Message{}, "", llm.Usage{}, err
|
||||||
}
|
}
|
||||||
@@ -264,8 +346,28 @@ func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionReque
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
if callErr == nil {
|
if callErr == nil {
|
||||||
|
resetLLMFailStreak(taskID, selection.profileID)
|
||||||
return msg, sr, usage, nil
|
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.
|
// profileID=0: the explicit chain was cleared; non-quota error: pass through. Neither changes the task failover state.
|
||||||
if selection.profileID == 0 || !isQuotaExhaustedError(callErr) {
|
if selection.profileID == 0 || !isQuotaExhaustedError(callErr) {
|
||||||
return llm.Message{}, "", llm.Usage{}, callErr
|
return llm.Message{}, "", llm.Usage{}, callErr
|
||||||
@@ -287,7 +389,7 @@ func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionReque
|
|||||||
func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) iter.Seq2[llm.StreamEvent, error] {
|
func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) iter.Seq2[llm.StreamEvent, error] {
|
||||||
return func(yield func(llm.StreamEvent, error) bool) {
|
return func(yield func(llm.StreamEvent, error) bool) {
|
||||||
for {
|
for {
|
||||||
selection, err := hooks.current()
|
selection, err := hooks.current(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
yield(llm.StreamEvent{}, err)
|
yield(llm.StreamEvent{}, err)
|
||||||
return
|
return
|
||||||
@@ -338,6 +440,7 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
if streamErr == nil {
|
if streamErr == nil {
|
||||||
|
resetLLMFailStreak(taskID, selection.profileID)
|
||||||
for _, buffered := range pending {
|
for _, buffered := range pending {
|
||||||
if !yield(buffered, nil) {
|
if !yield(buffered, nil) {
|
||||||
return
|
return
|
||||||
@@ -345,6 +448,28 @@ func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest
|
|||||||
}
|
}
|
||||||
return
|
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
|
// profileID=0 means the explicit chain was cleared while this stable task
|
||||||
// bundle was still in use. Agent/global fallback errors follow the legacy
|
// bundle was still in use. Agent/global fallback errors follow the legacy
|
||||||
// behavior and never mutate task failover state.
|
// behavior and never mutate task failover state.
|
||||||
|
|||||||
@@ -0,0 +1,80 @@
|
|||||||
|
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) {
|
||||||
|
resetLLMFailStreaksForTest()
|
||||||
|
prov := &failingProvider{err: errors.New("boom: upstream 500")}
|
||||||
|
profile := int64(7)
|
||||||
|
nextProfile := int64(8)
|
||||||
|
exhaustCalls := 0
|
||||||
|
hooks := taskLLMStreamHooks{
|
||||||
|
current: func(ctx context.Context) (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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@@ -0,0 +1,32 @@
|
|||||||
|
package server
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestLLMProfilePinRoundTrip(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
if _, ok := llmProfilePinFrom(ctx); ok {
|
||||||
|
t.Fatal("empty context must carry no pin")
|
||||||
|
}
|
||||||
|
pinned := withLLMProfilePin(ctx, 42)
|
||||||
|
got, ok := llmProfilePinFrom(pinned)
|
||||||
|
if !ok || got != 42 {
|
||||||
|
t.Fatalf("pin roundtrip got=%d ok=%v", got, ok)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNextRotatedProfileRoundRobin(t *testing.T) {
|
||||||
|
tk := &Task{LLMProfileIDs: []int64{3, 1, 2}}
|
||||||
|
want := []int64{3, 1, 2, 3, 1}
|
||||||
|
for i, w := range want {
|
||||||
|
got, ok := tk.nextRotatedProfile()
|
||||||
|
if !ok || got != w {
|
||||||
|
t.Fatalf("rotation %d: got=%d ok=%v want=%d", i, got, ok, w)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if _, ok := (&Task{}).nextRotatedProfile(); ok {
|
||||||
|
t.Fatal("empty chain must not rotate")
|
||||||
|
}
|
||||||
|
}
|
||||||
+11
-10
@@ -246,7 +246,7 @@ func TestTaskLLMStreamRetriesPreStreamQuotaOnNextProfile(t *testing.T) {
|
|||||||
active := int64(11)
|
active := int64(11)
|
||||||
transitions := 0
|
transitions := 0
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
if active == 11 {
|
if active == 11 {
|
||||||
return taskLLMSelection{profileID: 11, provider: first}, nil
|
return taskLLMSelection{profileID: 11, provider: first}, nil
|
||||||
}
|
}
|
||||||
@@ -288,7 +288,7 @@ func TestTaskLLMStreamRetriesAfterNonContentStartEvent(t *testing.T) {
|
|||||||
}}
|
}}
|
||||||
active := int64(11)
|
active := int64(11)
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
if active == 11 {
|
if active == 11 {
|
||||||
return taskLLMSelection{profileID: 11, provider: first}, nil
|
return taskLLMSelection{profileID: 11, provider: first}, nil
|
||||||
}
|
}
|
||||||
@@ -319,7 +319,7 @@ func TestTaskLLMStreamRetriesStalePreStreamFailureAgainstReplacementChain(t *tes
|
|||||||
second := &scriptedLLMProvider{events: []llm.StreamEvent{{Type: llm.SETextDelta, Text: "replacement"}}}
|
second := &scriptedLLMProvider{events: []llm.StreamEvent{{Type: llm.SETextDelta, Text: "replacement"}}}
|
||||||
current := first
|
current := first
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
profileID := int64(11)
|
profileID := int64(11)
|
||||||
if current == second {
|
if current == second {
|
||||||
profileID = 22
|
profileID = 22
|
||||||
@@ -350,10 +350,11 @@ func TestTaskLLMStreamDoesNotSwitchForOrdinaryErrors(t *testing.T) {
|
|||||||
}
|
}
|
||||||
for _, message := range tests {
|
for _, message := range tests {
|
||||||
t.Run(message, func(t *testing.T) {
|
t.Run(message, func(t *testing.T) {
|
||||||
|
resetLLMFailStreaksForTest() // package-global streaks must not leak across subtests
|
||||||
provider := &scriptedLLMProvider{err: errors.New(message)}
|
provider := &scriptedLLMProvider{err: errors.New(message)}
|
||||||
exhausted := 0
|
exhausted := 0
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
||||||
},
|
},
|
||||||
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
||||||
@@ -382,7 +383,7 @@ func TestTaskLLMStreamAdvancesAfterMidStreamQuotaWithoutReplay(t *testing.T) {
|
|||||||
second := &scriptedLLMProvider{events: []llm.StreamEvent{{Type: llm.SETextDelta, Text: "next-call"}}}
|
second := &scriptedLLMProvider{events: []llm.StreamEvent{{Type: llm.SETextDelta, Text: "next-call"}}}
|
||||||
active := int64(11)
|
active := int64(11)
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
if active == 11 {
|
if active == 11 {
|
||||||
return taskLLMSelection{profileID: 11, provider: first}, nil
|
return taskLLMSelection{profileID: 11, provider: first}, nil
|
||||||
}
|
}
|
||||||
@@ -418,7 +419,7 @@ func TestTaskLLMStreamMarkFailureAfterToolStartSkipsWorkerReplay(t *testing.T) {
|
|||||||
err: quota,
|
err: quota,
|
||||||
}
|
}
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
||||||
},
|
},
|
||||||
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
||||||
@@ -467,7 +468,7 @@ func TestTaskLLMStreamRetriesTransientPreCommitOnSameProvider(t *testing.T) {
|
|||||||
}
|
}
|
||||||
exhausted := 0
|
exhausted := 0
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
||||||
},
|
},
|
||||||
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
||||||
@@ -499,7 +500,7 @@ func TestTaskLLMStreamSurfacesTransientAfterRetriesExhausted(t *testing.T) {
|
|||||||
failErr: errors.New("dial tcp: connection reset by peer"),
|
failErr: errors.New("dial tcp: connection reset by peer"),
|
||||||
}
|
}
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
||||||
},
|
},
|
||||||
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
||||||
@@ -524,7 +525,7 @@ func TestTaskLLMStreamDoesNotRetryDeterministicRejection(t *testing.T) {
|
|||||||
failErr: errors.New("anthropic: status 400: messages.1: invalid request"),
|
failErr: errors.New("anthropic: status 400: messages.1: invalid request"),
|
||||||
}
|
}
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
||||||
},
|
},
|
||||||
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
||||||
@@ -620,7 +621,7 @@ func TestTaskLLMStreamStopsWhenChainExhausted(t *testing.T) {
|
|||||||
provider := &scriptedLLMProvider{err: errors.New("billing_not_active")}
|
provider := &scriptedLLMProvider{err: errors.New("billing_not_active")}
|
||||||
transitions := 0
|
transitions := 0
|
||||||
hooks := taskLLMStreamHooks{
|
hooks := taskLLMStreamHooks{
|
||||||
current: func() (taskLLMSelection, error) {
|
current: func(ctx context.Context) (taskLLMSelection, error) {
|
||||||
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
return taskLLMSelection{profileID: 11, provider: provider}, nil
|
||||||
},
|
},
|
||||||
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
exhaust: func(taskLLMSelection, error) (db.TaskLLMTransition, error) {
|
||||||
|
|||||||
@@ -478,7 +478,7 @@ function WorkerAssetBadge({ assets }: { assets: IntentAsset[] }) {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
export function SessionsTab({ taskId }: { taskId: string }) {
|
export function SessionsTab({ taskId, taskPaused = false }: { taskId: string; taskPaused?: boolean }) {
|
||||||
const approvalFocus = useApprovalFocus({ taskId });
|
const approvalFocus = useApprovalFocus({ taskId });
|
||||||
const [selectedSessionId, setActiveId] = React.useState(MAIN_ID);
|
const [selectedSessionId, setActiveId] = React.useState(MAIN_ID);
|
||||||
const focusSession = React.useMemo<Session | undefined>(() => {
|
const focusSession = React.useMemo<Session | undefined>(() => {
|
||||||
@@ -1416,6 +1416,9 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
|||||||
|
|
||||||
// Run duration = span from this session's first step to its last (for a live
|
// Run duration = span from this session's first step to its last (for a live
|
||||||
// session, "now" so it ticks up — the 1.5s setTick above re-renders it).
|
// session, "now" so it ticks up — the 1.5s setTick above re-renders it).
|
||||||
|
// While the TASK is paused nothing time-derived may advance (#84): sessions
|
||||||
|
// keep state='running' in the store during a pause, so live alone would keep
|
||||||
|
// ticking and misrepresent the frozen budget. Freeze at the last step instead.
|
||||||
const runDuration = React.useMemo(() => {
|
const runDuration = React.useMemo(() => {
|
||||||
let min = Infinity,
|
let min = Infinity,
|
||||||
max = 0;
|
max = 0;
|
||||||
@@ -1426,9 +1429,9 @@ export function SessionsTab({ taskId }: { taskId: string }) {
|
|||||||
if (t > max) max = t;
|
if (t > max) max = t;
|
||||||
}
|
}
|
||||||
if (!Number.isFinite(min) || max === 0) return null;
|
if (!Number.isFinite(min) || max === 0) return null;
|
||||||
const end = active.live ? Date.now() : max;
|
const end = active.live && !taskPaused ? Date.now() : max;
|
||||||
return Math.max(0, end - min);
|
return Math.max(0, end - min);
|
||||||
}, [activity, active.live]);
|
}, [activity, active.live, taskPaused]);
|
||||||
|
|
||||||
// ---- transcript auto-scroll (open → bottom; stick to bottom unless scrolled up) ----
|
// ---- transcript auto-scroll (open → bottom; stick to bottom unless scrolled up) ----
|
||||||
const contentRef = React.useRef<HTMLDivElement | null>(null);
|
const contentRef = React.useRef<HTMLDivElement | null>(null);
|
||||||
|
|||||||
@@ -420,7 +420,7 @@ function TaskDetailInner() {
|
|||||||
{/* Tab content */}
|
{/* Tab content */}
|
||||||
<div className="flex-1 p-4 lg:p-6">
|
<div className="flex-1 p-4 lg:p-6">
|
||||||
<TabsContent value="sessions" className="mt-0">
|
<TabsContent value="sessions" className="mt-0">
|
||||||
<SessionsTab taskId={id} />
|
<SessionsTab taskId={id} taskPaused={task?.status === "paused"} />
|
||||||
</TabsContent>
|
</TabsContent>
|
||||||
<TabsContent value="overview" className="mt-0">
|
<TabsContent value="overview" className="mt-0">
|
||||||
<OverviewTab taskId={id} />
|
<OverviewTab taskId={id} />
|
||||||
|
|||||||
Reference in New Issue
Block a user