Files
ARTEX/server/server.go
T
BaaRai 07d272727d fix: emit safe skill frontmatter YAML and verify it round-trips
Descriptions written through the API were interpolated raw into the SKILL.md
frontmatter, so a value containing a colon+space made the whole frontmatter
unparseable: every field silently stayed empty, the skill loaded under its
directory name, and a skill-gated MCP server was silently skipped. The same
applied to updates through the meta rewrite, and the shipped scopesentry skill
had a malformed name line all along.

- yamlScalar: plain-safe tokens stay bare, everything else single-quoted
- fsCreateSkill/fsUpdateSkillMeta emit safe values and re-verify the written
  file with the real skill parser (name + mcps must survive), rejecting the
  write otherwise
- warnBrokenSkills at startup logs skills whose name fell back to the
  directory (the signature of broken frontmatter)
- fix skills/scopesentry/SKILL.md name line

Closes #39
2026-10-11 22:26:22 +02:00

3938 lines
148 KiB
Go

package server
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net"
"net/http"
"net/url"
"os"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/Autumn-27/artex/agent"
"github.com/Autumn-27/artex/db"
"github.com/Autumn-27/artex/intercept"
"github.com/Autumn-27/artex/llmpool"
"github.com/Autumn-27/artex/llmrec"
"github.com/Autumn-27/artex/report"
"github.com/Autumn-27/artex/traffic"
"github.com/Autumn-27/norma/llm"
actool "github.com/Autumn-27/norma/tool"
"github.com/Autumn-27/norma/transcript"
)
// BuildVersion is the backend application version, injected from cmd/artex at
// startup (which in turn gets it from -ldflags "-X main.version=<tag>").
// Defaults to "dev" for local builds. Exposed to the frontend via GET /api/health.
var BuildVersion = "dev"
// Server exposes the ARTEX backend over a JSON HTTP API for the shadcn/ui
// frontend.
type Server struct {
m *Manager
engine *Engine
ctx context.Context
skillDir string // root directory for skill subdirectories
jwtKey []byte // HS256 signing key loaded from / generated into dataDir/jwt.key
// concMu serializes concurrency-cap decisions (admission + reconcile) so a
// scheduler tick and an HTTP settings change / task creation can't both count
// free slots off the same snapshot and over-promote past the limit.
concMu sync.Mutex
cfgMu sync.Mutex
chatAgent *agent.ChatAgent // conversational runner for the chat page; nil w/o LLM
// llmProv is the fully-decorated global provider (recorder + failover chain)
// installed by applyLLM. Task routers use it after an explicit chain is cleared.
llmProv llm.Provider
llmDirect llm.Provider // concrete global provider, before any failover pool
llmCfg agent.Config // current LLM config (key not exposed)
llmOn bool
llmProf string // active LLM profile name (for llmrec tagging)
// chatBusy guards the per-task main-agent run: the chat handler launches the
// agent on the server's background ctx (not the request ctx) and returns
// immediately, so a page reload / proxy timeout can't abort a live run. This
// map serializes turns per task — one main-agent run at a time per task, so
// concurrent messages don't corrupt the shared exp<id>-main transcript.
chatMu sync.Mutex
chatBusy map[string]bool
// chatCancel holds the cancel func for each in-flight conversation run (keyed by
// convBusyKey), so a manual stop can abort JUST that session's agent run. Set/
// cleared alongside chatBusy under chatMu. Aborting a run does NOT touch the P3
// trigger queue — the drain goroutine simply proceeds to the next queued fire.
chatCancel map[string]context.CancelCauseFunc
// triggerQ buffers P3 trigger fires PER AGENT. A per-agent "pump" launches runs up
// to a concurrency limit derived from the agent's policy: serial → limit 1 (+ optional
// merge); parallel → limit = trigger_max_parallel (0=∞), no merge. triggerActive
// counts in-flight runs per agent (replaces a boolean drain flag); a run's
// completion decrements it and re-pumps to fill the freed slot. triggerCfg caches
// the agent's last-read policy so the pump never queries the DB while holding queueMu.
// Distinct agents always run concurrently. Queue is in-memory (matches chatBusy); a
// restart drops pending fires — the scheduler re-fires from watermarks next tick.
queueMu sync.Mutex
triggerQ map[string][]triggeredRun
triggerActive map[string]int
triggerCfg map[string]triggerBehavior
// profChatAgents caches a per-profile ChatAgent (chat page), keyed by profile id.
// Built lazily on first use; invalidated when any profile is saved/activated/deleted
// so edits take effect.
profMu sync.Mutex
profChatAgents map[int64]*agent.ChatAgent // per-profile ChatAgent cache (chat page)
// provByProfile caches ONE provider per LLM profile id so every agent bound or
// pinned to the same profile shares a single provider instance — hence one rate
// limiter. applyLLM also resolves the persisted global-active profile through this
// cache, so global fallback and task chains do not accidentally double the
// configured request rate. Cleared on profile edits, then repopulated by active reapply.
provCacheMu sync.Mutex
provByProfile map[int64]*provEntry
provCacheGen uint64
// llmHealth is the process-wide circuit-breaker state for LLM failover (failover pool).
// It deliberately lives OUTSIDE the provider caches: rebuilding the chain
// (saving an unrelated profile, flipping a setting) must not erase what we
// learned about which backends are out of credit / rate-limited.
llmHealth *llmpool.Registry
// Explicit task chains use one task-scoped dynamic provider shared by goals,
// planner, workers, and the main agent. Bundles are stable; their runtime reads
// the persisted current profile at every new LLM call.
taskAgentMu sync.Mutex
taskAgents map[string]*taskAgentBundle
// Cold task archives run through one persistent FIFO worker. The buffered wake
// channel coalesces enqueue bursts; the database remains the source of truth.
archiveWake chan struct{}
archiveWG sync.WaitGroup
side *sideQuestionState
}
// provEntry is a cached provider + its config for one LLM profile id.
type provEntry struct {
prov llm.Provider
cfg agent.Config
}
// triggeredRun is one queued P3 trigger fire awaiting its turn for an agent.
// taskID + mergeable let the drainer coalesce several event triggers (finding/goal)
// from the SAME task into one conversation before it starts (interval fires don't merge).
type triggeredRun struct {
agentKey string
title string
message string // event body (trigger phrase + tool/args/results etc.); no task description/goal header
taskID int64 // source task for finding/goal triggers; 0 for interval/none
taskDesc string // task description (task-level, same across one task); rendered once when merged
taskGoal string // task goal (task-level, same across one task); rendered once when merged
mergeable bool // true for finding/goal event triggers (merge by taskID)
}
func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDir string) *Server {
key, err := loadOrCreateJWTKey(keyDir, dataDir)
if err != nil {
log.Fatalf("[auth] JWT key: %v", err)
}
s := &Server{m: m, engine: NewEngine(m), ctx: ctx, skillDir: skillDir, jwtKey: key, chatBusy: map[string]bool{},
chatCancel: map[string]context.CancelCauseFunc{}, triggerQ: map[string][]triggeredRun{},
triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{},
profChatAgents: map[int64]*agent.ChatAgent{},
provByProfile: map[int64]*provEntry{}, llmHealth: newLLMHealthRegistry(m.pg),
taskAgents: map[string]*taskAgentBundle{}, archiveWake: make(chan struct{}, 1)}
s.initSideQuestions()
// Circuit-breaker thresholds/cooldowns are hot parameters on the failure path; push the global retry policy to the Registry once at startup,
// and again on every policy save (saveLLMRetryPolicy).
s.applyRetryPolicy()
// Every task uses a stable task router. An empty explicit chain is resolved by
// that router through Agent bindings and then the global provider, so adding a
// first chain to a running task takes effect on its very next LLM call.
s.engine.SetAuthoritativeAgentResolver(func(t *Task) (*agent.Planner, *agent.Worker) {
if !s.taskRuntimeAvailable(t, "planner", "worker") {
return nil, nil
}
b := s.agentsForTask(t)
return b.pl, b.wk
})
// Global readiness (Ready()/llm_configured): a global active LLM provider is
// installed. Task-level runnability is separate (ReadyFor → the resolver above).
s.engine.SetReadiness(func() bool {
s.cfgMu.Lock()
defer s.cfgMu.Unlock()
return s.llmOn
})
// Wire DB-stored prompt templates into the agents (new design plan §3.3 / §5a). With no
// override row, agents keep their built-in defaults — behavior is unchanged.
if m.pg != nil {
agent.PromptOverride = func(key string) (string, bool) {
a, err := m.pg.GetAgentByKey(key)
if err != nil || a == nil {
return "", false
}
t, err := m.pg.CurrentPrompt(a.ID)
if err != nil || t == "" {
return "", false
}
return t, true
}
// Wire DB-stored wrap-up (settlement) prompts. Empty column → built-in default.
agent.WrapupOverride = func(key string) (string, bool) {
a, err := m.pg.GetAgentByKey(key)
if err != nil || a == nil || a.WrapupPrompt == "" {
return "", false
}
return a.WrapupPrompt, true
}
// Wire DB-stored wrap-up turn budgets. 0 / missing → built-in per-agent default.
agent.WrapupMaxTurnsOverride = func(key string) (int, bool) {
a, err := m.pg.GetAgentByKey(key)
if err != nil || a == nil || a.WrapupMaxTurns <= 0 {
return 0, false
}
return a.WrapupMaxTurns, true
}
// Wire DB-stored TASK-TIMEOUT wrap-up prompt/turns (worker/planner). Empty/0 → default.
agent.WrapupTaskTimeoutOverride = func(key string) (string, bool) {
a, err := m.pg.GetAgentByKey(key)
if err != nil || a == nil || a.TaskTimeoutWrapupPrompt == "" {
return "", false
}
return a.TaskTimeoutWrapupPrompt, true
}
agent.WrapupTaskTimeoutTurnsOverride = func(key string) (int, bool) {
a, err := m.pg.GetAgentByKey(key)
if err != nil || a == nil || a.TaskTimeoutWrapupMaxTurns <= 0 {
return 0, false
}
return a.TaskTimeoutWrapupMaxTurns, true
}
wireAgentAugment(m.pg, s.skillDir, s.hostTools) // wire visible skills/MCP + traffic/orchestration host tools into the agent toolset
domainReg := buildDomainReg(m.Assets())
wireTools(m.pg, domainReg) // built-in tools table: filter per agent + override description/schema + inject defaults
seedPrompts(m.pg) // seed built-in agent default prompt bodies into agent_prompts (only when empty)
warnBrokenSkills(s.skillDir) // surface skills whose frontmatter silently failed to parse
s.seedOrchestrationTools() // seed P2 cross-task orchestration tools into the tools table (bindable per agent)
if err := s.seedFindingRetester(); err != nil {
log.Printf("[retester] seed: %v", err)
}
go s.evidenceStore().RunGC(s.ctx)
s.seedPythonInterpreter() // custom script tool: detect the python interpreter at boot and store it (only when empty)
go newScheduler(s).Run(s.ctx) // P3 trigger scheduling (scheduled/finding/goal events), custom agents only
// Vulnerability IM push delivery engine. Sits beside the Scheduler but independent: push latency demands (3s)
// differ from the triggers' business cadence, and neither's failures drag the other down — a stuck push must not affect agent triggers.
go newNotifier(s).Run(s.ctx)
// Fill the tool cache for any enabled MCP that has none yet (notably the
// seeded browser MCP on first run). Async so it never blocks startup.
go s.discoverEmptyMCPsOnStartup()
logSink.SetDB(ctx, m.pg) // restore last 100 log rows and enable async persistence
}
// precedence: persisted DB config > env.
if cfg, ok := s.loadLLMConfig(); ok {
if err := s.applyLLM(cfg); err != nil {
log.Printf("[engine] saved LLM config init failed — engine idle: %v", err)
} else {
log.Printf("[engine] LLM configured from DB: %s / %s", cfg.Provider(), cfg.Model)
}
} else if cfg, ok := agent.FromEnv(); ok {
if err := s.applyLLM(cfg); err != nil {
log.Printf("[engine] env provider init failed — engine idle: %v", err)
} else {
log.Printf("[engine] LLM configured from env: %s / %s", cfg.Provider(), cfg.Model)
}
} else {
log.Printf("[engine] no LLM provider configured — engine idle until set via /api/llm or env")
}
s.restoreTaskRuntimes()
go s.reconcileConcurrency()
s.startTaskArchiveWorker()
s.wireInterceptReviewer() // LLM fallback approval: commands that hit no interception rule go to the model for judgment
return s
}
// Restored deadline and worker loops must inherit the same context as new tasks,
// including the side-question checkpoint publisher installed during startup.
func (s *Server) restoreTaskRuntimes() {
m := s.m
// reload tasks persisted on disk so the task list survives a restart, and
// restore persisted paused state (so a task paused before restart stays paused).
for _, t := range m.LoadExisting() {
lifecycle := t.lifecycleSnapshot()
// clear stale 'running' intents from a prior crash/restart (no live worker
// owns them) so they re-claim instead of spinning forever in the UI.
if n, _ := t.Store.ResetRunningIntents(); n > 0 {
log.Printf("[engine] task %s reset %d leftover running intents to open", t.ID, n)
}
if lifecycle.Paused {
s.engine.Pause(t.ID, agent.AbortPausedOnReload)
}
// Task-level timeout: start a deadline coordinator for every non-terminal task with a timeout, independent of the planner/worker
// loop — inactive tasks also get wrapped up on time after a restart (an already-passed deadline runs the wrap-up sequence immediately).
if !isTerminalStatus(lifecycle.Status) {
s.engine.startDeadlineCoordinator(s.ctx, t)
}
}
// Restore every task that had already been admitted before shutdown. Starting
// only the active UI task left other non-queued tasks counted as concurrency
// occupants without live loops, which could permanently block the persistent
// FIFO. Paused loops remain idle; queued tasks are admitted below as slots allow.
for _, t := range m.List() {
lifecycle := t.lifecycleSnapshot()
if !lifecycle.Queued && !isTerminalStatus(lifecycle.Status) {
s.engine.Run(s.ctx, t)
}
}
}
// agentMaxTurns returns the configured max_turns for an agent key (0 = unlimited,
// also the fallback when no DB or no row).
func (s *Server) agentMaxTurns(key string) int {
if s.m.pg == nil {
return 0
}
a, err := s.m.pg.GetAgentByKey(key)
if err != nil || a == nil {
return 0
}
return a.MaxTurns
}
// agentRunSeconds returns the configured wall-clock run budget (seconds) for an
// agent key (0 = unlimited; 1200 fallback when no DB or no row, matching schema).
func (s *Server) agentRunSeconds(key string) int {
if s.m.pg == nil {
return 1200
}
a, err := s.m.pg.GetAgentByKey(key)
if err != nil || a == nil {
return 1200
}
return a.RunSecs
}
// loadLLMConfig reads the active LLM profile from PG (llm_profiles).
func (s *Server) loadLLMConfig() (agent.Config, bool) {
p, err := s.m.pg.ActiveProfile()
if err != nil || p == nil {
return agent.Config{}, false
}
cfg := agent.ConfigFrom(p.Format, p.Model, p.BaseURL, p.APIKey, p.Proxy)
cfg.RatePerSecond, cfg.RatePerMinute = p.RatePerSecond, p.RatePerMinute
cfg.ContextWindowK = p.ContextWindowK
cfg.ThinkingType = p.ThinkingType
cfg.ReasoningEffort = p.ReasoningEffort
cfg.Stream = p.Streaming
cfg.MaxTokens, cfg.MaxTokensField = p.MaxTokens, p.MaxTokensField
cfg.SessionHeaderKey = p.SessionHeaderKey
s.applyProfileRetry(&cfg, p)
if cfg.APIKey == "" {
return cfg, false
}
s.cfgMu.Lock()
s.llmProf = p.Name
s.cfgMu.Unlock()
return cfg, true
}
// saveLLMConfig persists the LLM config as the active "default" profile in PG.
func (s *Server) saveLLMConfig(cfg agent.Config) error {
// cfg.Provider() already returns one of the three valid format strings
// (anthropic / openai / openai-responses), matching the DB CHECK constraint.
format := cfg.Provider()
var id int64
// This legacy endpoint's request body carries no failover/streaming/output-cap params, so the values already stored in the DB are passed back untouched —
// otherwise every save would silently reset the profile's priority, pool_exclude, streaming and output caps
// (max_tokens / max_tokens_field) to zero values.
var priority int
var poolExclude bool
streaming := true // legacy DBs / new saves default to streaming
var maxTokens int
var maxTokensField string
var sessionHeaderKey string
var retry db.RetryOverride
if profs, _ := s.m.pg.ListProfiles(); profs != nil {
for _, p := range profs {
if p.Name == "default" {
id, priority, poolExclude, streaming = p.ID, p.Priority, p.PoolExclude, p.Streaming
maxTokens, maxTokensField = p.MaxTokens, p.MaxTokensField
sessionHeaderKey = p.SessionHeaderKey
retry = p.Retry
break
}
}
}
newID, err := s.m.pg.SaveProfile(&db.LLMProfile{
ID: id, Name: "default", Format: format, Model: cfg.Model, BaseURL: cfg.BaseURL, Proxy: cfg.Proxy,
APIKey: cfg.APIKey, RatePerSecond: cfg.RatePerSecond, RatePerMinute: cfg.RatePerMinute,
ContextWindowK: cfg.ContextWindowK, ThinkingType: cfg.ThinkingType, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true,
Priority: priority, PoolExclude: poolExclude, Streaming: streaming,
MaxTokens: maxTokens, MaxTokensField: maxTokensField, SessionHeaderKey: sessionHeaderKey,
Retry: retry,
})
if err != nil {
return err
}
return s.m.pg.SetActiveProfile(newID)
}
// reapplyActiveProfile hot-reloads the engine from the active DB profile so that
// saving or activating a profile takes effect without a restart. Best-effort:
// logs on failure and leaves the running engine untouched.
func (s *Server) reapplyActiveProfile() {
cfg, ok := s.loadLLMConfig()
if !ok {
return
}
if err := s.applyLLM(cfg); err != nil {
log.Printf("[engine] reapply active profile failed: %v", err)
return
}
log.Printf("[engine] LLM reapplied from active profile: %s / %s", cfg.Provider(), cfg.Model)
}
// webSearchFor gates the global web-search opts (backend/key) by an agent's own
// web_search flag: the backend/key come from the global config, each agent decides on/off.
func (s *Server) webSearchFor(key string) agent.WebSearchOpts {
o := s.m.WebSearchOpts()
if a, err := s.m.pg.GetAgentByKey(key); err != nil || a == nil || !a.WebSearch {
o.Enabled = false
}
return o
}
// nonStreamingResolver returns a resolver capturing a profile's streaming choice.
// The agents read it per run; a profile change rebuilds the agents (applyLLM),
// so the captured value is always the one in effect for this build.
func nonStreamingResolver(cfg agent.Config) func() bool {
nonStreaming := !cfg.Stream
return func() bool { return nonStreaming }
}
// maxTokensResolver mirrors nonStreamingResolver for the per-reply output cap.
func maxTokensResolver(cfg agent.Config) func() int {
maxTokens := cfg.MaxTokens
return func() int { return maxTokens }
}
// applyLLM (re)builds the planner/worker/main-agent from cfg and installs them on
// the running engine as the GLOBAL active pair. Safe to call at runtime (UI configures LLM).
func (s *Server) applyLLM(cfg agent.Config) error {
var prov llm.Provider
// A persisted active profile must use the same cached provider as task chains
// and Agent bindings, otherwise each path owns a separate rate limiter.
if active, _ := s.m.pg.ActiveProfile(); active != nil {
if activeCfg, ok := s.loadProfileConfig(active.ID); ok && activeCfg == cfg {
prov, _, _ = s.providerForProfile(active.ID)
}
}
if prov == nil {
var err error
prov, err = cfg.NewProvider()
if err != nil {
return err
}
// Non-persisted/env configs do not have a profile cache key.
s.cfgMu.Lock()
profName := s.llmProf
s.cfgMu.Unlock()
prov = llmrec.Wrap(prov, s.m.PG(), cfg.Model, profName, cfg.ThinkingType, cfg.ReasoningEffort, s.m.LLMRecordEnabled)
prov = bindSideProvider(prov, cfg, 0, profName)
}
s.cfgMu.Lock()
s.llmDirect = prov
s.cfgMu.Unlock()
// LLM failover pool (default off): wraps the active profile in a failover chain and switches to the next one automatically when the current is unavailable.
// It only affects this "global active profile" path — agent-bound / task-pinned ones go through providerForProfile,
// still owning that profile exclusively by default (see poolForBinding). When off or with no alternatives, the original provider is returned and behavior is unchanged.
if act, err := s.m.pg.ActiveProfile(); err == nil && act != nil {
prov = s.poolForActive(act.ID, prov, cfg)
}
// No global planner/worker pair: every task runs on its own task-routed pair
// (agentsForTask), resolved through the engine's authoritative resolver. Global
// readiness (Ready()/llm_configured) is reported from s.llmOn, set below.
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
win := cfg.CompactionWindow()
// chat stays the GLOBAL fallback since one ChatAgent serves many agent keys — its
// per-agent binding is resolved at Chat time (runConversationSync →
// chatAgentForProfile). The main-agent has no global instance: each task builds its
// own via agentsForTask (task-routed), so nothing is constructed for it here.
s.cfgMu.Lock()
// chat agent serves MANY custom agents by key → it holds the GLOBAL opts
// (backend/key) and gates Enabled per-conversation-agent at Chat time. Conversations always use the active profile.
s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner
s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
s.chatAgent.SetWebSearch(s.m.WebSearchOpts())
s.chatAgent.SetGuard(s.chatGuard())
s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg))
s.chatAgent.SetMaxTokens(maxTokensResolver(cfg))
s.chatAgent.SetNoaEnabled(s.m.NoaCompactionEnabled) // experimental: noa context compaction (read per run)
s.llmProv = prov
s.llmCfg = cfg
s.llmOn = true
s.cfgMu.Unlock()
s.invalidateTaskAgents()
// wake the active task so a task created while idle starts exploring.
if t := s.m.ActiveTask(); t != nil {
t.Notify()
}
return nil
}
// loadProfileConfig builds an agent.Config from a specific profile id (with its key).
// ok=false when the profile is missing or has no api key.
func (s *Server) loadProfileConfig(id int64) (agent.Config, bool) {
p, err := s.m.pg.ProfileByID(id)
if err != nil || p == nil {
return agent.Config{}, false
}
cfg := agent.ConfigFrom(p.Format, p.Model, p.BaseURL, p.APIKey, p.Proxy)
cfg.RatePerSecond, cfg.RatePerMinute = p.RatePerSecond, p.RatePerMinute
cfg.ContextWindowK = p.ContextWindowK
cfg.ThinkingType = p.ThinkingType
cfg.ReasoningEffort = p.ReasoningEffort
cfg.Stream = p.Streaming
cfg.MaxTokens, cfg.MaxTokensField = p.MaxTokens, p.MaxTokensField
cfg.SessionHeaderKey = p.SessionHeaderKey
s.applyProfileRetry(&cfg, p)
if cfg.APIKey == "" {
return cfg, false
}
return cfg, true
}
// effectiveProfileForAgent resolves the LLM profile id an agent should run on, by
// precedence: agent binding (agents.llm_profile_id) → pin (task/conversation) → nil
// (caller falls back to the global active profile). A binding to a deleted profile
// can't happen (FK ON DELETE SET NULL); an otherwise-invalid one is dropped downstream
// by loadProfileConfig, letting the caller fall back.
func (s *Server) effectiveProfileForAgent(agentKey string, pinID *int64) *int64 {
if s.m.pg != nil && agentKey != "" {
if a, _ := s.m.pg.GetAgentByKey(agentKey); a != nil && a.LLMProfileID != nil {
return a.LLMProfileID
}
}
return pinID
}
// resolveChatAgent picks the ChatAgent for one conversation: the agent's own
// binding or this conversation's chosen profile first, the global active config
// only as a fallback. Both the send precheck and the background runner MUST use
// this — resolving differently in the two paths is how a conversation that had
// picked a valid profile still got rejected with "LLM not configured" when no global
// config was active.
func (s *Server) resolveChatAgent(c *db.Conversation) *agent.ChatAgent {
ca := s.chatAgentRef()
if eff := s.effectiveProfileForAgent(c.AgentKey, c.LLMProfileID); eff != nil {
if pa := s.chatAgentForProfile(*eff); pa != nil {
ca = pa
}
}
return ca
}
// chatUnavailableReason explains why no ChatAgent could be resolved, so the user
// knows whether to add a config, activate one, or pick one for this conversation
// — rather than a flat "not configured" that hides which of those it is.
func (s *Server) chatUnavailableReason() string {
if s.m.pg != nil {
if profiles, err := s.m.pg.ListProfiles(); err == nil && len(profiles) == 0 {
return "No LLM configured yet: go to System → LLM Profiles to add one"
}
if active, err := s.m.pg.ActiveProfile(); err == nil && active == nil {
return "No active LLM profile: activate one under System → LLM Profiles, or pick one for this conversation"
}
}
return "LLM not ready for chat: check that System → LLM Profiles has a usable, activated profile"
}
// providerForProfile returns a cached provider+cfg for a profile id, so every agent
// bound/pinned to the same profile shares one provider instance (one rate limiter).
// ok=false when the profile is missing/invalid → caller falls back to the global pair.
func (s *Server) providerForProfile(id int64) (llm.Provider, agent.Config, bool) {
s.provCacheMu.Lock()
if e := s.provByProfile[id]; e != nil {
s.provCacheMu.Unlock()
return e.prov, e.cfg, true
}
generation := s.provCacheGen
s.provCacheMu.Unlock()
cfg, ok := s.loadProfileConfig(id)
if !ok {
return nil, agent.Config{}, false
}
prov, err := cfg.NewProvider()
if err != nil {
log.Printf("[engine] build provider for LLM profile %d failed: %v", id, err)
return nil, agent.Config{}, false
}
// Wrap with recorder, tagged with this profile's name.
if p, _ := s.m.pg.ProfileByID(id); p != nil {
prov = llmrec.Wrap(prov, s.m.PG(), cfg.Model, p.Name, cfg.ThinkingType, cfg.ReasoningEffort, s.m.LLMRecordEnabled)
prov = bindSideProvider(prov, cfg, id, p.Name)
}
s.provCacheMu.Lock()
if generation != s.provCacheGen {
s.provCacheMu.Unlock()
return s.providerForProfile(id)
}
if e := s.provByProfile[id]; e != nil { // lost the race → keep the winner
prov, cfg = e.prov, e.cfg
} else {
s.provByProfile[id] = &provEntry{prov: prov, cfg: cfg}
log.Printf("[engine] built provider for LLM profile %d (%s / %s)", id, cfg.Provider(), cfg.Model)
}
s.provCacheMu.Unlock()
return prov, cfg, true
}
// chatAgentForProfile returns a ChatAgent built from a specific LLM profile, cached
// per profile id. Returns nil if the profile is missing or has no API key.
func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
s.profMu.Lock()
cached := s.profChatAgents[id]
s.profMu.Unlock()
if cached != nil {
return cached
}
prov, cfg, ok := s.providerForProfile(id)
if !ok {
return nil
}
tx := transcript.NewStore(filepath.Join(s.m.dir, "transcripts"))
ca := agent.NewChatAgent(s.poolForBinding(id, prov, cfg), cfg.Model, s.m.dir, tx, cfg.CompactionWindow())
ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
ca.SetWebSearch(s.m.WebSearchOpts())
ca.SetGuard(s.chatGuard())
ca.SetNonStreaming(nonStreamingResolver(cfg))
ca.SetMaxTokens(maxTokensResolver(cfg))
ca.SetNoaEnabled(s.m.NoaCompactionEnabled) // experimental: noa context compaction (read per run)
s.profMu.Lock()
if ex := s.profChatAgents[id]; ex != nil { // lost the race → keep the winner
ca = ex
} else {
s.profChatAgents[id] = ca
}
s.profMu.Unlock()
return ca
}
// invalidateProfileAgents drops the per-profile ChatAgent + provider caches so a profile
// save/activate/delete — or an agent's binding change — rebuilds pinned tasks' agents
// (and re-resolves each agent's bound model) on their next round.
func (s *Server) invalidateProfileAgents() {
s.profMu.Lock()
s.profChatAgents = map[int64]*agent.ChatAgent{}
s.profMu.Unlock()
s.provCacheMu.Lock()
s.provCacheGen++
s.provByProfile = map[int64]*provEntry{}
s.provCacheMu.Unlock()
s.invalidateTaskAgents()
}
func (s *Server) chatAgentRef() *agent.ChatAgent {
s.cfgMu.Lock()
defer s.cfgMu.Unlock()
return s.chatAgent
}
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
s.registerSideRoutes(mux)
// Auth routes — exempt from JWT check (handled in requireAuth)
mux.HandleFunc("GET /api/auth/status", s.authStatus)
mux.HandleFunc("POST /api/auth/init", s.authInit)
mux.HandleFunc("POST /api/auth/login", s.authLogin)
mux.HandleFunc("POST /api/auth/change-password", s.authChangePassword)
mux.HandleFunc("GET /api/health", s.health)
mux.HandleFunc("GET /api/stats", s.stats)
mux.HandleFunc("GET /api/logs", s.getLogs)
mux.HandleFunc("GET /api/logs/history", s.getLogsHistory)
mux.HandleFunc("GET /api/logs/stream", s.streamLogs)
// One-click page update. These go through the default JWT auth (auth.go only exempts /api/auth/* and
// /api/health), so the endpoints that modify the program itself naturally require a login.
mux.HandleFunc("GET /api/update/check", s.updateCheck)
mux.HandleFunc("POST /api/update/apply", s.updateApply)
mux.HandleFunc("POST /api/update/rollback", s.updateRollback)
mux.HandleFunc("GET /api/update/stream", s.updateStream)
mux.HandleFunc("GET /api/tasks", s.listTasks)
mux.HandleFunc("POST /api/tasks", s.createTask)
mux.HandleFunc("GET /api/task-categories", s.pgListTaskCategories)
mux.HandleFunc("POST /api/task-categories", s.pgCreateTaskCategory)
mux.HandleFunc("PATCH /api/task-categories/{id}", s.pgRenameTaskCategory)
mux.HandleFunc("DELETE /api/task-categories/{id}", s.pgDeleteTaskCategory)
mux.HandleFunc("POST /api/tasks/category/batch", s.updateTasksCategoryBatch)
mux.HandleFunc("GET /api/task-templates", s.pgListTaskTemplates)
mux.HandleFunc("POST /api/task-templates", s.pgCreateTaskTemplate)
mux.HandleFunc("PATCH /api/task-templates/{id}", s.pgUpdateTaskTemplate)
mux.HandleFunc("DELETE /api/task-templates/{id}", s.pgDeleteTaskTemplate)
mux.HandleFunc("GET /api/tasks/{id}", s.getTask)
mux.HandleFunc("PATCH /api/tasks/{id}", s.updateTaskMetadata)
mux.HandleFunc("PATCH /api/tasks/{id}/category", s.updateTaskCategory)
// task-level asset interception/allow rules
mux.HandleFunc("GET /api/tasks/{id}/intercept-rules", s.taskInterceptListRules)
mux.HandleFunc("POST /api/tasks/{id}/intercept-rules", s.taskInterceptCreateRule)
mux.HandleFunc("PUT /api/tasks/{id}/intercept-rules/{rid}", s.taskInterceptUpdateRule)
mux.HandleFunc("DELETE /api/tasks/{id}/intercept-rules/{rid}", s.taskInterceptDeleteRule)
mux.HandleFunc("POST /api/tasks/{id}/intercept-rules/{rid}/toggle", s.taskInterceptToggleRule)
mux.HandleFunc("POST /api/tasks/control/batch", s.controlTasksBatch)
mux.HandleFunc("GET /api/task-archives", s.listTaskArchives)
mux.HandleFunc("GET /api/task-archives/{id}", s.getTaskArchive)
mux.HandleFunc("GET /api/tasks/{id}/credentials", s.listTaskCredentials)
mux.HandleFunc("POST /api/tasks/{id}/credentials", s.addTaskCredential)
mux.HandleFunc("DELETE /api/tasks/{id}/credentials/{cid}", s.deleteTaskCredential)
mux.HandleFunc("POST /api/tasks/{id}/archive", s.queueTaskArchive)
mux.HandleFunc("POST /api/tasks/archive/batch", s.queueTaskArchivesBatch)
mux.HandleFunc("POST /api/task-archives/{id}/restore", s.queueTaskArchiveRestore)
mux.HandleFunc("POST /api/task-archives/restore/batch", s.restoreTaskArchivesBatch)
mux.HandleFunc("DELETE /api/task-archives/{id}", s.queueTaskArchiveDelete)
mux.HandleFunc("POST /api/task-archives/delete/batch", s.deleteTaskArchivesBatch)
mux.HandleFunc("GET /api/tasks/{id}/coverage", s.taskCoverage)
mux.HandleFunc("GET /api/tasks/{id}/coverage-graph", s.taskCoverageGraph)
mux.HandleFunc("GET /api/tasks/{id}/asset-refs", s.taskAssetRefs)
mux.HandleFunc("POST /api/tasks/{id}/assets", s.attachTaskAssets)
mux.HandleFunc("DELETE /api/tasks/{id}/assets/{assetID}", s.detachTaskAsset)
mux.HandleFunc("GET /api/tasks/{id}/intent-assets", s.taskIntentAssets)
// workspace file manager (over workDir)
mux.HandleFunc("GET /api/workspace/list", s.wsList)
mux.HandleFunc("GET /api/workspace/read", s.wsRead)
mux.HandleFunc("POST /api/workspace/write", s.wsWrite)
mux.HandleFunc("POST /api/workspace/mkdir", s.wsMkdir)
mux.HandleFunc("DELETE /api/workspace/delete", s.wsDelete)
mux.HandleFunc("GET /api/workspace/download", s.wsDownload)
mux.HandleFunc("POST /api/workspace/upload", s.wsUpload)
mux.HandleFunc("GET /api/tasks/{id}/scope", s.taskScopeList)
mux.HandleFunc("POST /api/tasks/{id}/scope", s.taskScopeAdd)
mux.HandleFunc("DELETE /api/tasks/{id}/scope/{sid}", s.taskScopeDelete)
mux.HandleFunc("GET /api/tasks/{id}/goals", s.listGoals) // goal management: list all goals of this task
mux.HandleFunc("POST /api/tasks/{id}/goals", s.addGoal) // goal management: human-added goal (revives the task)
mux.HandleFunc("PATCH /api/tasks/{id}/goals/{gid}", s.editGoal) // goal management: edit a goal (revives the task)
mux.HandleFunc("DELETE /api/tasks/{id}/goals/{gid}", s.deleteGoal) // goal management: hard-delete a goal (no revival)
mux.HandleFunc("GET /api/tasks/{id}/constraints", s.listConstraints) // constraint management: list this task's operation constraints
mux.HandleFunc("POST /api/tasks/{id}/constraints", s.addConstraint) // constraint management: add a constraint (planner not notified)
mux.HandleFunc("PATCH /api/tasks/{id}/constraints/{cid}", s.editConstraint) // constraint management: edit a constraint
mux.HandleFunc("DELETE /api/tasks/{id}/constraints/{cid}", s.deleteConstraint) // constraint management: delete a constraint
mux.HandleFunc("POST /api/tasks/{id}/control", s.control)
mux.HandleFunc("PUT /api/tasks/{id}/llm", s.updateTaskLLMProfiles)
mux.HandleFunc("GET /api/tasks/{id}/llm/resolution", s.taskLLMResolutionHandler)
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/control", s.controlIntent)
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/messages", s.sendWorkerMessage)
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/rerun", s.rerunIntent) // rerun one blocked/exhausted/stopped intent
mux.HandleFunc("POST /api/tasks/{id}/intents/rerun-blocked", s.rerunBlocked) // batch-rerun all blocked intents of this task
mux.HandleFunc("POST /api/active", s.setActive)
mux.HandleFunc("GET /api/llm", s.getLLM)
mux.HandleFunc("POST /api/llm", s.setLLM)
mux.HandleFunc("POST /api/llm/test", s.testLLM)
// asset system
mux.HandleFunc("GET /api/assets", s.listAssets)
mux.HandleFunc("GET /api/assets/counts", s.assetCounts)
mux.HandleFunc("POST /api/assets", s.insertAssets)
mux.HandleFunc("DELETE /api/assets", s.deleteAssets)
// company system
mux.HandleFunc("GET /api/companies", s.listCompanies)
mux.HandleFunc("POST /api/companies", s.createCompany)
mux.HandleFunc("GET /api/companies/{id}", s.getCompany)
mux.HandleFunc("DELETE /api/companies/{id}", s.deleteCompany)
mux.HandleFunc("POST /api/companies/{id}/scope", s.addCompanyScope)
mux.HandleFunc("POST /api/companies/reattribute", s.reattribute)
mux.HandleFunc("GET /api/exploration/frontier", s.frontier)
mux.HandleFunc("GET /api/exploration/findings", s.findings)
mux.HandleFunc("GET /api/exploration/findings/groups", s.findingGroups)
mux.HandleFunc("GET /api/exploration/findings/asset-tree", s.findingAssetTree)
mux.HandleFunc("GET /api/exploration/findings/stats", s.findingStats)
mux.HandleFunc("GET /api/exploration/findings/export", s.findingsExport)
s.registerFindingTraffic(mux)
mux.HandleFunc("GET /api/exploration/findings/{id}", s.getFinding)
mux.HandleFunc("GET /api/exploration/findings/{id}/lineage", s.findingLineage)
mux.HandleFunc("POST /api/exploration/findings/{id}/deepen", s.deepenFinding)
mux.HandleFunc("GET /api/exploration/findings/{id}/retests", s.listFindingRetests)
mux.HandleFunc("GET /api/exploration/findings/retests/active", s.listActiveFindingRetests)
mux.HandleFunc("POST /api/exploration/findings/{id}/retests", s.startFindingRetest)
mux.HandleFunc("PATCH /api/exploration/findings/{id}", s.patchFinding)
mux.HandleFunc("DELETE /api/exploration/findings/{id}", s.deleteFinding)
mux.HandleFunc("GET /api/exploration/intents", s.intents)
mux.HandleFunc("GET /api/exploration/graph", s.explorationGraph)
mux.HandleFunc("GET /api/exploration/nodes", s.explorationNodes)
mux.HandleFunc("GET /api/exploration/activity", s.activity)
mux.HandleFunc("GET /api/exploration/activity/history", s.activityHistory)
mux.HandleFunc("GET /api/exploration/main-sessions", s.mainSessions)
mux.HandleFunc("POST /api/exploration/main-session/new", s.newMainSession)
mux.HandleFunc("GET /api/exploration/activity/stream", s.streamActivity)
mux.HandleFunc("GET /api/exploration/activity/{seq}", s.activityDetail)
mux.HandleFunc("GET /api/exploration/tokens", s.tokenStats)
mux.HandleFunc("GET /api/tokens/daily", s.tokenDailyStats)
mux.HandleFunc("GET /api/tokens/conversations", s.conversationTokens)
mux.HandleFunc("GET /api/tokens/usage", s.pgUsageStats) // global llm_usage aggregation (new dashboard view)
mux.HandleFunc("GET /api/audit", s.getAudit)
mux.HandleFunc("POST /api/gc", s.gc)
mux.HandleFunc("GET /api/traffic", s.getTraffic)
mux.HandleFunc("GET /api/traffic/hosts", s.getTrafficHosts)
mux.HandleFunc("DELETE /api/traffic", s.deleteTraffic)
mux.HandleFunc("DELETE /api/traffic/hosts", s.deleteTrafficHosts)
mux.HandleFunc("DELETE /api/traffic/all", s.deleteAllTraffic)
mux.HandleFunc("GET /api/traffic/exchange", s.getTrafficExchange)
mux.HandleFunc("GET /api/traffic/blob", s.getTrafficBlob)
mux.HandleFunc("POST /api/traffic/ingest", s.ingestTraffic)
mux.HandleFunc("GET /api/commands", s.pgListCommands)
mux.HandleFunc("GET /api/commands/stats", s.pgToolStats) // aggregate call counts per tool
mux.HandleFunc("GET /api/llm/records", s.pgListLLMRecords)
mux.HandleFunc("DELETE /api/llm/records", s.pgDeleteLLMRecords)
mux.HandleFunc("GET /api/llm/records/tasks", s.pgLLMTasks)
mux.HandleFunc("GET /api/llm/records/by-model", s.pgTokenByModel) // aggregate this task's token usage by model
mux.HandleFunc("GET /api/llm/records/{id}", s.pgGetLLMRecord)
mux.HandleFunc("GET /api/settings", s.getSettings)
mux.HandleFunc("PUT /api/settings", s.putSettings)
// Vulnerability IM push. Channels are "multi-instance + per-instance filter rules" resources, so they get their own group of
// REST endpoints instead of being stuffed into the flat /api/settings key-value store.
mux.HandleFunc("GET /api/notify/meta", s.notifyMeta)
mux.HandleFunc("GET /api/notify/channels", s.notifyListChannels)
mux.HandleFunc("POST /api/notify/channels", s.notifyCreateChannel)
mux.HandleFunc("PATCH /api/notify/channels/{id}", s.notifyUpdateChannel)
mux.HandleFunc("DELETE /api/notify/channels/{id}", s.notifyDeleteChannel)
mux.HandleFunc("POST /api/notify/channels/{id}/test", s.notifyTestChannel)
mux.HandleFunc("GET /api/notify/deliveries", s.notifyListDeliveries)
mux.HandleFunc("POST /api/notify/deliveries/{id}/retry", s.notifyRetryDelivery)
mux.HandleFunc("POST /api/settings/web-search/test", s.testWebSearch)
mux.HandleFunc("GET /api/report", s.getReport)
mux.HandleFunc("GET /api/chat/mentions", s.searchChatMentions)
mux.HandleFunc("POST /api/chat", s.chat)
mux.HandleFunc("POST /api/chat/upload", s.chatUpload) // upload method 1: files land in uploads/ under the conversation/task work dir
mux.HandleFunc("GET /api/tasks/{id}/chat/status", s.taskChatStatus)
mux.HandleFunc("POST /api/tasks/{id}/chat/stop", s.stopChat)
// --- Admin API (PostgreSQL data source; new database & admin console plan) ---
mux.HandleFunc("DELETE /api/tasks/{id}", s.pgDeleteTask)
// Agents
// conversations (chat page)
mux.HandleFunc("GET /api/conversations", s.pgListConversations)
mux.HandleFunc("POST /api/conversations", s.pgCreateConversation)
mux.HandleFunc("POST /api/conversations/delete/batch", s.pgDeleteConversationsBatch)
mux.HandleFunc("PATCH /api/conversations/{id}", s.pgRenameConversation)
mux.HandleFunc("PATCH /api/conversations/{id}/profile", s.pgUpdateConversation)
mux.HandleFunc("DELETE /api/conversations/{id}", s.pgDeleteConversation)
mux.HandleFunc("GET /api/conversations/{id}/messages", s.pgConversationMessages)
mux.HandleFunc("POST /api/conversations/{id}/messages", s.pgSendConversationMessage)
mux.HandleFunc("POST /api/conversations/{id}/stop", s.pgStopConversation)
mux.HandleFunc("GET /api/conversations/{id}/messages/{seq}", s.pgConversationMsgDetail)
mux.HandleFunc("GET /api/agents", s.pgListAgents)
mux.HandleFunc("POST /api/agents", s.pgCreateAgent)
mux.HandleFunc("GET /api/agents/{key}", s.pgGetAgent)
mux.HandleFunc("PATCH /api/agents/{key}", s.pgUpdateAgent)
mux.HandleFunc("DELETE /api/agents/{key}", s.pgDeleteAgent)
mux.HandleFunc("PUT /api/agents/{key}/config", s.pgSaveAgentConfig)
mux.HandleFunc("PUT /api/agents/{key}/prompt", s.pgSavePrompt)
mux.HandleFunc("POST /api/agents/{key}/prompt/reset", s.pgResetPrompt)
mux.HandleFunc("PUT /api/agents/{key}/wrapup", s.pgSaveWrapup)
mux.HandleFunc("POST /api/agents/{key}/wrapup/reset", s.pgResetWrapup)
mux.HandleFunc("PUT /api/agents/{key}/wrapup/task-timeout", s.pgSaveTaskTimeoutWrapup)
mux.HandleFunc("POST /api/agents/{key}/wrapup/task-timeout/reset", s.pgResetTaskTimeoutWrapup)
mux.HandleFunc("GET /api/agents/{key}/triggers", s.pgListTriggers)
mux.HandleFunc("POST /api/agents/{key}/triggers", s.pgCreateTrigger)
mux.HandleFunc("PATCH /api/triggers/{id}", s.pgUpdateTrigger)
mux.HandleFunc("DELETE /api/triggers/{id}", s.pgDeleteTrigger)
mux.HandleFunc("GET /api/agents/{key}/prompts", s.pgListPromptVersions)
mux.HandleFunc("GET /api/agents/{key}/variables", s.pgPromptVars)
mux.HandleFunc("POST /api/agents/{key}/prompt/preview", s.pgPreviewPrompt)
mux.HandleFunc("GET /api/agents/{key}/visibility", s.pgGetAgentVisibility)
mux.HandleFunc("PUT /api/agents/{key}/visibility", s.pgSetAgentVisibility)
// built-in tool catalog (description/param defaults editable, bound per agent; key and handler live in code)
mux.HandleFunc("GET /api/tools", s.pgListTools)
mux.HandleFunc("PUT /api/tools/{key}", s.pgUpdateTool)
mux.HandleFunc("POST /api/tools/custom", s.pgCreateCustomTool)
mux.HandleFunc("POST /api/tools/custom/test", s.pgTestCustomTool)
mux.HandleFunc("PUT /api/tools/custom/{key}", s.pgUpdateCustomTool)
mux.HandleFunc("DELETE /api/tools/custom/{key}", s.pgDeleteCustomTool)
mux.HandleFunc("POST /api/settings/python/detect", s.pgDetectPython)
mux.HandleFunc("POST /api/tools/{key}/reset", s.pgResetTool)
// MCP CRUD
mux.HandleFunc("GET /api/mcp", s.pgListMCP)
mux.HandleFunc("POST /api/mcp", s.pgSaveMCP)
mux.HandleFunc("DELETE /api/mcp/{id}", s.pgDeleteMCP)
mux.HandleFunc("GET /api/mcp/{id}/tools", s.pgMCPTools)
mux.HandleFunc("POST /api/mcp/{id}/refresh", s.pgRefreshMCP)
// asset sync — ScopeSentry data source
mux.HandleFunc("GET /api/sync/scopesentry/status", s.syncSSStatus)
mux.HandleFunc("POST /api/sync/scopesentry/datasource", s.syncSSDatasource)
mux.HandleFunc("GET /api/sync/scopesentry/projects", s.syncSSProjects)
mux.HandleFunc("GET /api/sync/scopesentry/tasks", s.syncSSTasks)
mux.HandleFunc("POST /api/sync/scopesentry/sync", s.syncSSRun)
// Skill CRUD (filesystem)
mux.HandleFunc("GET /api/skills", s.fsListSkills)
mux.HandleFunc("POST /api/skills", s.fsCreateSkill)
mux.HandleFunc("POST /api/skills/upload", s.fsUploadSkill)
mux.HandleFunc("DELETE /api/skills/{name}", s.fsDeleteSkill)
mux.HandleFunc("GET /api/skills/missing", s.fsMissingSkills) // names of skills that were invoked but do not exist
mux.HandleFunc("GET /api/skills/{name}/usage", s.fsSkillUsage) // recent invocations of one skill
mux.HandleFunc("PUT /api/skills/{name}/meta", s.fsUpdateSkillMeta)
mux.HandleFunc("POST /api/skills/{name}/dirs", s.fsCreateDir)
mux.HandleFunc("GET /api/skills/{name}/files", s.fsListFiles)
// {file...} captures path segments including slashes (e.g. scripts/extract.py)
mux.HandleFunc("GET /api/skills/{name}/files/{file...}", s.fsReadFile)
mux.HandleFunc("PUT /api/skills/{name}/files/{file...}", s.fsWriteFile)
mux.HandleFunc("DELETE /api/skills/{name}/files/{file...}", s.fsDeletePath)
// MCP resource-side visibility (more specific skill routes match first)
mux.HandleFunc("GET /api/visibility/{kind}/{id}", s.pgResourceVisibility)
mux.HandleFunc("POST /api/visibility/toggle", s.pgToggleVisibility)
// Skill visibility (by name, more specific, takes priority over the wildcard routes above)
mux.HandleFunc("GET /api/visibility/skill/{name}", s.pgSkillVisibility)
mux.HandleFunc("POST /api/visibility/skill/toggle", s.pgToggleSkillVisibility)
// LLM multi-profile
mux.HandleFunc("GET /api/llm/profiles", s.pgListProfiles)
mux.HandleFunc("POST /api/llm/profiles", s.pgSaveProfile)
mux.HandleFunc("DELETE /api/llm/profiles/{id}", s.pgDeleteProfile)
mux.HandleFunc("POST /api/llm/profiles/active", s.pgActivateProfile)
mux.HandleFunc("GET /api/llm/retry-policy", s.pgGetLLMRetryPolicy)
mux.HandleFunc("POST /api/llm/retry-policy", s.pgSaveLLMRetryPolicy)
mux.HandleFunc("GET /api/llm/pool", s.pgLLMPoolStatus)
mux.HandleFunc("POST /api/llm/pool/reset", s.pgLLMPoolReset)
mux.HandleFunc("POST /api/llm/models", s.pgListModels)
// interception rule management
mux.HandleFunc("GET /api/intercept/rules", s.interceptListRules)
mux.HandleFunc("POST /api/intercept/rules", s.interceptCreateRule)
mux.HandleFunc("PUT /api/intercept/rules/{id}", s.interceptUpdateRule)
mux.HandleFunc("DELETE /api/intercept/rules/{id}", s.interceptDeleteRule)
mux.HandleFunc("POST /api/intercept/rules/{id}/toggle", s.interceptToggleRule)
// asset interception rule management (global blocklist: domains/IPs/URLs/CIDRs)
mux.HandleFunc("GET /api/asset-intercept/rules", s.assetInterceptListRules)
mux.HandleFunc("POST /api/asset-intercept/rules", s.assetInterceptCreateRule)
mux.HandleFunc("PUT /api/asset-intercept/rules/{id}", s.assetInterceptUpdateRule)
mux.HandleFunc("DELETE /api/asset-intercept/rules/{id}", s.assetInterceptDeleteRule)
mux.HandleFunc("POST /api/asset-intercept/rules/{id}/toggle", s.assetInterceptToggleRule)
mux.HandleFunc("GET /api/intercept/pending", s.interceptListPending)
mux.HandleFunc("GET /api/intercept/pending/{id}", s.interceptGetOne)
mux.HandleFunc("POST /api/intercept/pending/{id}/decide", s.interceptDecide)
mux.HandleFunc("GET /api/intercept/history", s.interceptHistory)
mux.HandleFunc("GET /api/intercept/history/{id}", s.interceptDetail)
mux.HandleFunc("GET /api/intercept/history/{id}/execution", s.interceptExecution)
mux.HandleFunc("GET /api/intercept/task/{taskID}", s.interceptListTaskItems)
mux.HandleFunc("GET /api/intercept/tool-config", s.interceptGetToolConfig)
mux.HandleFunc("PUT /api/intercept/tool-config", s.interceptSetToolConfig)
mux.HandleFunc("GET /api/intercept/judge", s.interceptGetJudgeConfig)
mux.HandleFunc("PUT /api/intercept/judge", s.interceptSetJudgeConfig)
mux.HandleFunc("GET /api/intercept/judge/usage", s.interceptJudgeUsage) // cumulative token usage of fallback approval
// /api/* goes through CORS + JWT; everything else is served by the embedded
// frontend (public — auth is enforced client-side and on the API). With the
// no-embed build the webui handler just 404s (run `next dev` separately).
api := cors(s.requireAuth(mux))
root := http.NewServeMux()
root.Handle("/api/", api)
root.Handle("/", s.webuiHandler())
return root
}
// --- handlers ---
func (s *Server) health(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"ok": true, "service": "artex", "version": BuildVersion})
}
func (s *Server) stats(w http.ResponseWriter, r *http.Request) {
out := map[string]any{}
out["engine_mode"] = "idle" // spec enum; overridden below when a task is active
out["llm_configured"] = s.engine.Ready() // is an LLM provider installed at all
if tr := s.m.Traffic(); tr != nil {
c, _ := tr.Count()
out["traffic"] = c
out["traffic_enabled"] = true
}
if counts, err := s.m.Assets().CountsByType(); err == nil {
total := 0
for _, n := range counts {
total += n
}
out["assets"] = total
out["asset_counts"] = counts
}
// resolve the task: explicit ?task=<id> binds to that task (so a detail view
// never silently follows a globally-changed active task); empty = active.
taskParam := r.URL.Query().Get("task")
t := s.m.ResolveTask(taskParam)
if t == nil {
if taskParam != "" && taskParam != "active" {
writeErr(w, 404, "task not found")
return
}
writeJSON(w, 200, out) // no task selected: only global fields
return
}
out["llm_configured"] = s.engine.ReadyFor(t)
st, _ := t.Store.Stats()
out["exploration"] = st
// per-task running state + heartbeat (distinct from "LLM configured").
intents, _ := t.Store.ListByKind(db.KindIntent, 100000)
inFlight := 0
for _, in := range intents {
if in.State == "running" {
inFlight++
}
}
goals, _ := t.Store.ListByKind(db.KindGoal, 10000)
goalsMet := 0
for _, g := range goals {
if g.State == "met" {
goalsMet++
}
}
last := s.engine.LastActivity(t.ID)
paused := s.engine.IsPaused(t.ID)
activeCalls := s.engine.ActiveLLMCalls(t.ID)
running := s.engine.ReadyFor(t) && s.engine.Started(t.ID) && !paused
// A live event-driven engine with no work is healthy idle. Only a persisted
// running intent without any corresponding LLM call can be considered stalled.
stalled := running && activeCalls == 0 && inFlight > 0 && last > 0 && time.Now().Unix()-last > 120
// engine_mode follows the spec enum (exploring|paused|stalled|idle).
engineMode := "idle"
switch {
case paused:
engineMode = "paused"
case stalled:
engineMode = "stalled"
case running && activeCalls > 0:
engineMode = "exploring"
}
out["engine_mode"] = engineMode
out["active_task"] = map[string]any{
"id": t.ID, "description": t.Description, "goal": t.Goal,
"running": running,
"paused": paused,
"engine_mode": engineMode,
"in_flight": inFlight,
"llm_in_flight": activeCalls,
"last_activity": last,
"stalled": stalled,
"goals_total": len(goals),
"goals_met": goalsMet,
}
writeJSON(w, 200, out)
}
func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) {
active := ""
if t := s.m.ActiveTask(); t != nil {
active = t.ID
}
list := s.m.List()
metrics, _ := s.m.PG().TaskListMetricsAll()
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
dtos := make([]TaskDTO, 0, len(list))
for _, t := range list {
dto := taskDTO(t, s.resolvedTaskStatus(t))
applyTaskArchiveBlocker(&dto, archiveBlockers)
metric := metrics[t.ExpID]
dto.Tokens = tokenTotalDTO(metric.Tokens)
// prefer the live in-memory heartbeat (fresher) and fall back to the
// persisted max activity time (survives restarts) for run-duration display.
dto.LastActivity = metric.LastActivity
if live := s.engine.LastActivity(t.ID); live > dto.LastActivity {
dto.LastActivity = live
}
dto.GoalsTotal = metric.Goals.Total
dto.GoalsMet = metric.Goals.Met
dto.InFlight = metric.RunningIntents
dto.Findings = FindingSeverityDTO{
Critical: metric.Findings.Critical,
High: metric.Findings.High,
Medium: metric.Findings.Medium,
Low: metric.Findings.Low,
}
dtos = append(dtos, dto)
}
writeJSON(w, 200, map[string]any{"tasks": dtos, "active": active})
}
func (s *Server) resolvedTaskStatus(t *Task) string {
if t == nil {
return "created"
}
lifecycle := t.lifecycleSnapshot()
switch {
case isTerminalStatus(lifecycle.Status):
return lifecycle.Status
case lifecycle.Queued:
return "queued"
case lifecycle.Paused || s.engine.IsPaused(t.ID):
return "paused"
case t.llmStateSnapshot().FailoverState == "chain_exhausted" && s.engine.Started(t.ID):
// LLM readiness is an execution dependency, not lifecycle state. Keeping
// an exhausted task running lets the user edit/reset its chain directly.
return "running"
case s.engine.ReadyFor(t) && s.engine.Started(t.ID):
return "running"
default:
return "created"
}
}
func (s *Server) setActive(w http.ResponseWriter, r *http.Request) {
var req struct {
ID string `json:"id"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
if !s.m.SetActive(req.ID) {
writeErr(w, 404, "task not found")
return
}
// resume the engine for the opened task (idempotent — no-op if already running).
// Queued tasks: only set active for viewing; the engine is not started (the concurrency cap holds, reconcile fills freed slots).
if t, ok := s.m.Task(req.ID); ok && !t.lifecycleSnapshot().Queued {
s.engine.Run(s.ctx, t)
}
writeJSON(w, 200, map[string]any{"active": req.ID})
}
// control pauses/resumes a task's autonomous execution (planner + workers).
func (s *Server) control(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
var req struct {
Action string `json:"action"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
if req.Action != "pause" && req.Action != "resume" {
writeErr(w, 400, "action must be pause|resume")
return
}
result, err := s.applyTaskControl(t, req.Action)
if err != nil {
writeErr(w, 409, err.Error())
return
}
writeJSON(w, 200, result)
}
// controlIntent pauses, resumes, or cancels one local worker intent. A running
// worker is always stopped before state mutation/cleanup, preventing tool output
// that arrives after the user action from recreating deleted blackboard records.
func (s *Server) controlIntent(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
if !s.engine.beginTaskOperation(t.ID) {
writeErr(w, 409, "task is being deleted; cannot control intents")
return
}
defer s.engine.decInflight(t.ID)
iid, err := strconv.ParseInt(r.PathValue("iid"), 10, 64)
if err != nil || iid <= 0 {
writeErr(w, 400, "bad intent id")
return
}
var req struct {
Action string `json:"action"`
Reason string `json:"reason"` // required for cancel (delete): deletion reason
Mode string `json:"mode"` // cancel only: soft (default, soft delete) | hard (real delete, cascades away exclusively-owned descendants)
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, "bad json: "+err.Error())
return
}
if req.Action != "pause" && req.Action != "resume" && req.Action != "cancel" {
writeErr(w, 400, "action must be pause|resume|cancel")
return
}
result, err := s.applyIntentControl(r.Context(), t, iid, req.Action, req.Reason, req.Mode)
if err != nil {
writeErr(w, 409, err.Error())
return
}
writeJSON(w, 200, result)
}
// rerunIntent re-runs an intent that did not succeed (blocked/exhausted/stopped): it is set back to open, the worker
// re-claims it and runs from scratch (facts/findings/assets already written back to the graph are kept); a terminal/paused task is revived as a side effect.
// Powers "click continue on an errored work" — one-click retry after network/LLM jitter caused a blocked state.
func restoreRerunIntent(t *Task, before *db.Node) error {
if t == nil || before == nil {
return fmt.Errorf("missing intent rollback snapshot")
}
if before.State == "blocked" && before.BlockedReason != "" {
return t.Store.SetIntentBlockedReason(before.ID, before.BlockedReason)
}
return t.Store.SetIntentState(before.ID, before.State)
}
func (s *Server) rerunIntent(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
iid, err := strconv.ParseInt(r.PathValue("iid"), 10, 64)
if err != nil {
writeErr(w, 400, "bad intent id")
return
}
if !s.engine.beginTaskOperation(t.ID) {
writeErr(w, 409, "task is being deleted; cannot rerun intents")
return
}
defer s.engine.decInflight(t.ID)
before, err := t.Store.GetNode(iid)
if err != nil {
writeErr(w, 500, err.Error())
return
}
reopened, err := t.Store.ReopenIntent(iid)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if !reopened {
writeErr(w, 409, "intent is not in a rerunnable state (only blocked/exhausted/stopped can be rerun)")
return
}
queued, err := s.admitTask(t, "resume")
if err != nil {
if rollbackErr := restoreRerunIntent(t, before); rollbackErr != nil {
err = fmt.Errorf("%w; restore intent %d after admission failure: %v", err, iid, rollbackErr)
}
writeErr(w, 500, err.Error())
return
}
log.Printf("[task] #%s intent #%d reopened (rerun)", t.ID, iid)
writeJSON(w, 200, map[string]any{"id": t.ID, "reopened": iid, "queued": queued})
}
// rerunBlocked batch-reruns every blocked intent of this task (handy for one-click retry of the many blocked
// intents one network/LLM disconnection leaves behind), setting them back to open and reviving the task; returns how many were reopened.
func (s *Server) rerunBlocked(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
if !s.engine.beginTaskOperation(t.ID) {
writeErr(w, 409, "task is being deleted; cannot rerun intents")
return
}
defer s.engine.decInflight(t.ID)
intents, err := t.Store.ListByKind(db.KindIntent, 1000000)
if err != nil {
writeErr(w, 500, err.Error())
return
}
before := make([]*db.Node, 0)
for _, intent := range intents {
if intent.State == "blocked" {
copy := *intent
before = append(before, &copy)
}
}
n, err := t.Store.ReopenBlockedIntents()
if err != nil {
writeErr(w, 500, err.Error())
return
}
queued := false
if n > 0 {
queued, err = s.admitTask(t, "resume")
if err != nil {
rollbackErrors := make([]string, 0)
for _, intent := range before {
if rollbackErr := restoreRerunIntent(t, intent); rollbackErr != nil {
rollbackErrors = append(rollbackErrors, fmt.Sprintf("intent %d: %v", intent.ID, rollbackErr))
}
}
if len(rollbackErrors) > 0 {
err = fmt.Errorf("%w; restore blocked intents after admission failure: %s", err, strings.Join(rollbackErrors, "; "))
}
writeErr(w, 500, err.Error())
return
}
log.Printf("[task] #%s batch-reopened %d blocked intents", t.ID, n)
}
writeJSON(w, 200, map[string]any{"id": t.ID, "reopened": n, "queued": queued})
}
// getLLM returns the current LLM config (key never exposed).
func (s *Server) getLLM(w http.ResponseWriter, r *http.Request) {
s.cfgMu.Lock()
defer s.cfgMu.Unlock()
writeJSON(w, 200, map[string]any{
"configured": s.llmOn,
"provider": s.llmCfg.Provider(),
"model": s.llmCfg.Model,
"base_url": s.llmCfg.BaseURL,
"proxy": s.llmCfg.Proxy,
"key_set": s.llmCfg.APIKey != "",
"rate_per_second": s.llmCfg.RatePerSecond,
"rate_per_minute": s.llmCfg.RatePerMinute,
"context_window_k": s.llmCfg.ContextWindowK,
"thinking_type": s.llmCfg.ThinkingType,
"reasoning_effort": s.llmCfg.ReasoningEffort,
})
}
// setLLM configures the LLM at runtime. A blank api_key keeps the existing key.
func (s *Server) setLLM(w http.ResponseWriter, r *http.Request) {
var req struct {
Provider string `json:"provider"`
Model string `json:"model"`
BaseURL string `json:"base_url"`
Proxy string `json:"proxy"`
APIKey string `json:"api_key"`
RatePerSecond float64 `json:"rate_per_second"`
RatePerMinute float64 `json:"rate_per_minute"`
ContextWindowK int `json:"context_window_k"`
ThinkingType string `json:"thinking_type"`
ReasoningEffort string `json:"reasoning_effort"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
cfg := agent.ConfigFrom(req.Provider, req.Model, req.BaseURL, req.APIKey, req.Proxy)
cfg.RatePerSecond, cfg.RatePerMinute = req.RatePerSecond, req.RatePerMinute
cfg.ThinkingType = req.ThinkingType
cfg.ReasoningEffort = req.ReasoningEffort
if k := req.ContextWindowK; k > 0 { // 0 = keep default (200K); cap at 1M
if k > 1000 {
k = 1000
}
cfg.ContextWindowK = k
}
if cfg.APIKey == "" {
s.cfgMu.Lock()
cfg.APIKey = s.llmCfg.APIKey // keep existing key if not re-entered
s.cfgMu.Unlock()
}
if cfg.APIKey == "" {
writeErr(w, 400, "api_key required")
return
}
// Validate provider construction before persisting it as the active profile.
if _, err := cfg.NewProvider(); err != nil {
writeErr(w, 400, "provider init failed: "+err.Error())
return
}
if err := s.saveLLMConfig(cfg); err != nil {
writeErr(w, 500, "persist provider failed: "+err.Error())
return
}
s.invalidateProfileAgents()
if _, ok := s.loadLLMConfig(); !ok {
writeErr(w, 500, "saved provider is unavailable")
return
}
if err := s.applyLLM(cfg); err != nil {
writeErr(w, 400, "provider init failed: "+err.Error())
return
}
log.Printf("[engine] LLM configured via UI: %s / %s", cfg.Provider(), cfg.Model)
s.getLLM(w, r)
}
// testLLM makes a real minimal completion to verify the config works.
func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) {
var req struct {
Provider string `json:"provider"`
Model string `json:"model"`
BaseURL string `json:"base_url"`
Proxy string `json:"proxy"`
APIKey string `json:"api_key"`
ThinkingType string `json:"thinking_type"`
ReasoningEffort string `json:"reasoning_effort"`
ProfileID *int64 `json:"profile_id"` // passed when testing a stored profile: if api_key is empty, its stored key is used
Streaming *bool `json:"streaming"` // omitted = streaming, the same defaults as when saving a profile
SessionHeaderKey string `json:"session_header_key"` // non-empty = the test also sends this custom session header, with a one-off random session id as the value
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
cfg := agent.ConfigFrom(req.Provider, req.Model, req.BaseURL, req.APIKey, req.Proxy)
// mirror production: send the SAME thinking params so a provider that rejects the
// reasoning_effort/thinking field fails the test too (no false "test ok, run 400").
cfg.ThinkingType = req.ThinkingType
cfg.ReasoningEffort = req.ReasoningEffort
// Likewise, the streaming mode follows that profile's choice: endpoints supporting only one of the two channels must fail
// here already, instead of discovering mid-conversation that "the profile that passed the test cannot run at all".
if req.Streaming != nil {
cfg.Stream = *req.Streaming
}
// The custom session header name follows that profile too: when non-empty, the test request also sends it (value is a one-off random session id,
// see TestConnection). Endpoints like opencode zen that mandate x-opencode-session return
// 400 without it, so the test path must send it too — otherwise "chat works, test 400s".
cfg.SessionHeaderKey = req.SessionHeaderKey
// API key resolution precedence: form input > the named profile's stored key > the global config's key.
// A stored profile's key is never sent back to the browser, so the form is empty when testing a stored profile and the key comes from the DB.
// Same for the session header name: when the form omits it, the stored profile's value is the fallback.
if req.ProfileID != nil && (cfg.APIKey == "" || cfg.SessionHeaderKey == "") {
if p, err := s.m.pg.ProfileByID(*req.ProfileID); err == nil && p != nil {
if cfg.APIKey == "" {
cfg.APIKey = p.APIKey
}
if cfg.SessionHeaderKey == "" {
cfg.SessionHeaderKey = p.SessionHeaderKey
}
}
}
if cfg.APIKey == "" {
s.cfgMu.Lock()
cfg.APIKey = s.llmCfg.APIKey
s.cfgMu.Unlock()
}
if cfg.APIKey == "" {
writeJSON(w, 200, map[string]any{"ok": false, "error": "no API key provided"})
return
}
// Retry params do [not] go into the connection test: the test has a 30s hard timeout, and stacking the configured retry counts/long intervals on top
// would only turn a perfectly usable endpoint into a "timeout failure". The test asks "does this endpoint work at all"; retry pacing
// is a runtime concern.
lat, reply, err := agent.TestConnection(r.Context(), cfg)
if err != nil {
writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error()})
return
}
// Return the model's actual reply so a "passed" test has evidence: you can see it really spoke, not just an HTTP 200.
writeJSON(w, 200, map[string]any{
"ok": true, "latency_ms": lat.Milliseconds(), "model": cfg.Model, "reply": truncateReply(reply),
})
}
// truncateReply clips a connection-test reply for display. A model told to answer
// "OK" can still ramble (or think out loud); the UI only needs enough to show it
// really said something. Rune-based so multibyte text never splits mid-character.
func truncateReply(s string) string {
const max = 200
r := []rune(s)
if len(r) <= max {
return s
}
return string(r[:max]) + "…"
}
type createTaskReq struct {
Name string `json:"name,omitempty"` // optional task name; omitted/empty = unnamed
CategoryID *int64 `json:"category_id,omitempty"`
Description string `json:"description"`
Goal string `json:"goal"`
LLMProfileID *int64 `json:"llm_profile_id,omitempty"` // the LLM profile to run this task on; omitted/null = use the active profile
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
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
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner heartbeat interval (seconds); 0/omitted = default 600 (10min); floor = default = 600, lower values are raised to 600
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).
CoverageEnabled *bool `json:"coverage_enabled,omitempty"` // asset coverage feature; omitted/null = default on (true). false = disables coverage calculation/display/auto-accumulated scope + hides add_task_scope/list_untested_assets. company links are unaffected.
// InterceptRules: task-level asset interception/allow rules (entered at creation, stored in task_intercept_rules, never the global table).
InterceptRules []taskInterceptRuleReq `json:"intercept_rules,omitempty"`
// Credentials: test accounts declared at creation (#3). An empty password means
// "agent generates": the agent registers that account and records it back.
Credentials []taskCredentialReq `json:"credentials,omitempty"`
}
type taskCredentialReq struct {
Domain string `json:"domain"`
Username string `json:"username"`
Password string `json:"password"`
Note string `json:"note"`
}
func (s *Server) createTask(w http.ResponseWriter, r *http.Request) {
var req createTaskReq
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, "bad json: "+err.Error())
return
}
if strings.TrimSpace(req.Description) == "" {
req.Description = "Unnamed task"
}
if len(req.LLMProfileIDs) == 0 && req.LLMProfileID != nil {
req.LLMProfileIDs = []int64{*req.LLMProfileID}
}
if err := s.validateTaskProfileIDs(req.LLMProfileIDs); err != nil {
writeErr(w, 400, err.Error())
return
}
if req.TimeoutSeconds < 0 {
req.TimeoutSeconds = 0
}
if len(req.SourceTaskIDs) > db.MaxTaskSourceCount {
writeErr(w, 400, fmt.Sprintf("at most %d source tasks may be selected", db.MaxTaskSourceCount))
return
}
sourceIDs := make([]int64, 0, len(req.SourceTaskIDs))
seenSources := map[int64]bool{}
for _, raw := range req.SourceTaskIDs {
id, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
if err != nil || id <= 0 || seenSources[id] {
writeErr(w, 400, "invalid or duplicate source task id")
return
}
if _, ok := s.m.Task(strconv.FormatInt(id, 10)); !ok {
writeErr(w, 400, fmt.Sprintf("source task #%d not found", id))
return
}
seenSources[id] = true
sourceIDs = append(sourceIDs, id)
}
companyIDs, err := db.NormalizeTaskCompanyIDs(req.CompanyIDs)
if err != nil {
writeErr(w, 400, fmt.Sprintf("invalid linked companies: at most %d valid companies may be selected", db.MaxTaskCompanyCount))
return
}
req.CompanyIDs = companyIDs
interceptRules, err := buildTaskInterceptRules(req.InterceptRules)
if err != nil {
writeErr(w, 400, "invalid task-level intercept rules: "+err.Error())
return
}
t, err := s.m.CreateTaskWithOptions(req.Description, req.Goal, db.TaskCreateOptions{
Name: strings.TrimSpace(req.Name), CategoryID: req.CategoryID,
SourceTaskIDs: sourceIDs, CompanyIDs: req.CompanyIDs, LLMProfileIDs: req.LLMProfileIDs,
TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds,
CoverageEnabled: req.CoverageEnabled,
InterceptRules: interceptRules,
})
if err != nil {
if errors.Is(err, db.ErrTaskCategoryInvalid) || errors.Is(err, db.ErrTaskCategoryNotFound) {
writeErr(w, 400, "task category not found or invalid")
return
}
if errors.Is(err, db.ErrTaskCompanyIDsInvalid) || errors.Is(err, db.ErrTaskCompanyNotFound) {
writeErr(w, 400, "linked company not found or invalid")
return
}
writeErr(w, 500, err.Error())
return
}
log.Printf("[task] created task #%s «%s» goal: %s", t.ID, req.Description, req.Goal)
// Task credentials (#3): rows land after the task row commits; a bad entry
// fails the whole request so the UI never shows a half-configured task.
if len(req.Credentials) > 0 {
taskNum, _ := strconv.ParseInt(t.ID, 10, 64)
for _, c := range req.Credentials {
if _, err := s.m.PG().UpsertTaskCredential(taskNum, db.TaskCredentialInput{
Domain: c.Domain, Username: c.Username, Password: c.Password,
Source: "user", Note: c.Note,
}); err != nil {
writeErr(w, 400, "invalid credential for "+c.Domain+": "+err.Error())
return
}
}
}
// Shared post-create flow (seed + seed intent + background goal decomposition + engine.Run), reusing the same launchTask as spawn_task.
// launchTask is async internally and never blocks the UI — goal decomposition visibly proceeds in the background.
s.launchTask(t, req.Description+" "+req.Goal, req.SeedFirstIntent != nil && *req.SeedFirstIntent)
writeJSON(w, 201, taskDTO(t, s.resolvedTaskStatus(t)))
}
func (s *Server) validateTaskProfileIDs(ids []int64) error {
seen := map[int64]bool{}
for _, id := range ids {
if id <= 0 || seen[id] {
return fmt.Errorf("invalid or duplicate LLM profile id")
}
seen[id] = true
if _, ok := s.loadProfileConfig(id); !ok {
return fmt.Errorf("LLM profile #%d not found or has no API key set", id)
}
}
return nil
}
func (s *Server) updateTaskLLMProfiles(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
// Any lifecycle state (including terminal) may edit the chain: after the task ends, main-agent conversations still run on it,
// and not switching when the model is unavailable would lock interactions of finished tasks too.
before := t.llmStateSnapshot()
var req struct {
LLMProfileIDs []int64 `json:"llm_profile_ids"`
ActiveLLMProfileID *int64 `json:"active_llm_profile_id"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, "bad json: "+err.Error())
return
}
if err := s.validateTaskProfileIDs(req.LLMProfileIDs); err != nil {
writeErr(w, 400, err.Error())
return
}
active := int64(0)
if req.ActiveLLMProfileID != nil {
active = *req.ActiveLLMProfileID
}
reopened, err := s.m.ReplaceTaskLLMProfiles(t.ID, req.LLMProfileIDs, active)
if err != nil {
writeErr(w, 400, err.Error())
return
}
// Profile edits affect only subsequent LLM calls. Existing in-flight calls
// retain their provider. Concurrency reconciliation treats ActiveLLMCalls as a
// live slot and postpones an unavailable task's pause/queue transition until
// the call returns; a nudge lets an idle running task resume promptly.
t.Notify()
go s.reconcileConcurrency()
llmState := t.llmStateSnapshot()
result := map[string]any{
"id": t.ID, "llm_profile_ids": llmState.ProfileIDs,
"active_llm_profile_id": llmState.ActiveID,
"llm_failover_state": llmState.FailoverState, "reopened_intents": reopened,
}
if !sameOptionalID(before.ActiveID, llmState.ActiveID) {
event := s.emitManualTaskLLMSwitch(t, before.ActiveID, llmState.ActiveID)
result["switch_event"] = activityDTO(event)
}
writeJSON(w, 200, result)
}
var (
reURL = regexp.MustCompile(`https?://[^\s'"]+`)
reIPPort = regexp.MustCompile(`\b((?:\d{1,3}\.){3}\d{1,3})(?::(\d{1,5}))?`)
reDomain = regexp.MustCompile(`\b((?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,})(?::(\d{1,5}))?`)
)
// parseTarget extracts a target (scheme, host, port) from free text, supporting
// full URLs, IP[:port] and domain[:port]. ok=false when nothing parseable.
func parseTarget(text string) (scheme, host string, port int, ok bool) {
text = strings.TrimSpace(text)
if m := reURL.FindString(text); m != "" {
if u, err := url.Parse(m); err == nil && u.Hostname() != "" {
scheme = strings.ToLower(u.Scheme)
host = strings.ToLower(u.Hostname())
port = portOr(u.Port(), defaultPort(scheme))
return scheme, host, port, true
}
}
if m := reIPPort.FindStringSubmatch(text); m != nil {
host = m[1]
port = portOr(m[2], 80)
return schemeForPort(port), host, port, true
}
if m := reDomain.FindStringSubmatch(text); m != nil {
host = strings.ToLower(m[1])
if m[2] == "" {
return "https", host, 443, true
}
port = portOr(m[2], 443)
return schemeForPort(port), host, port, true
}
return "", "", 0, false
}
func portOr(s string, d int) int {
if n, err := strconv.Atoi(s); err == nil && n > 0 {
return n
}
return d
}
func defaultPort(scheme string) int {
if scheme == "http" {
return 80
}
return 443
}
func schemeForPort(p int) string {
if p == 443 || p == 8443 {
return "https"
}
return "http"
}
// llmHost returns the host of the configured LLM endpoint (to keep it out of scope).
func (s *Server) llmHost() string {
s.cfgMu.Lock()
base := s.llmCfg.BaseURL
s.cfgMu.Unlock()
if base == "" {
return ""
}
if u, err := url.Parse(base); err == nil {
return strings.ToLower(u.Hostname())
}
return ""
}
func (s *Server) seed(t *Task, text string) {
scheme, host, port, ok := parseTarget(text)
if !ok {
log.Printf("[seed] task %s: could not parse a target host/IP from %q; no site created (configure scope manually)", t.ID, text)
return
}
// P0-1 guard: never treat the configured LLM gateway as a target.
if gw := s.llmHost(); gw != "" && host == gw {
log.Printf("[seed] task %s: target %q is the LLM gateway; refusing it as a penetration-testing target", t.ID, host)
return
}
u := scheme + "://" + host
if !(scheme == "https" && port == 443) && !(scheme == "http" && port == 80) {
u += ":" + strconv.Itoa(port)
}
var rootID int64
if as := s.m.Assets(); as != nil {
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
if net.ParseIP(host) != nil {
rootID, _ = as.UpsertIP(db.UpsertIPReq{IP: host, TaskID: taskID})
} else if scheme == "https" || scheme == "http" {
rootID, _ = as.UpsertHTTPService(db.UpsertHTTPServiceReq{URL: u, TaskID: taskID})
} else {
rootID, _ = as.UpsertRootDomain(db.UpsertRootDomainReq{Domain: host, TaskID: taskID})
}
if rootID > 0 {
_ = as.SetTaskAssetSource(taskID, rootID, "task", "initialized from the task description or goal", nil)
}
}
// anchor the seeded assets to this task's begin root as lineage/provenance
// (the asset graph is global and shared; anchoring no longer gates reads).
if rootID > 0 {
if begin, _ := t.Store.OriginFactID(); begin > 0 {
_ = t.Store.Anchor(begin, rootID)
}
}
log.Printf("[seed] task %s: target site %s", t.ID, u)
// No Notify here: whether the first round fires is decided solely by engine.Run's HasActiveIntent (seed-intent tasks
// skip the first round). seed runs before Run; a Notify here would buffer into the channel and be consumed when plannerLoop
// starts, bypassing Run's gate → seed tasks would wrongly fire a first round.
}
// seedFirstIntent writes ONE open intent (summary = description + goal) into the task's
// frontier at creation, so a worker can claim and run it immediately without first
// waiting a planner round. Mirrors a planner top-level intent: it links from the
// origin fact (RelDerivedFrom) so it still traces back to a fact node. Best-effort —
// a failure just falls back to the normal planner-driven flow.
func (s *Server) seedFirstIntent(t *Task) {
summary := fmt.Sprintf("Complete the task goal: %s (task: %s)", t.Goal, t.Description)
id, err := t.Store.AddIntent(map[string]any{"summary": summary}, 8, nil, "seed")
if err != nil {
log.Printf("[seed] task %s: failed to issue seed intent: %v", t.ID, err)
return
}
if origin, _ := t.Store.OriginFactID(); origin > 0 {
_ = t.Store.Link(origin, db.RelDerivedFrom, id)
}
log.Printf("[seed] task %s: seed intent #%d issued", t.ID, id)
}
func (s *Server) getTask(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
dto := taskDTO(t, s.resolvedTaskStatus(t))
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
applyTaskArchiveBlocker(&dto, archiveBlockers)
writeJSON(w, 200, dto)
}
// taskCoverage returns a task's rough asset test coverage (denominator/tested/backlog).
func (s *Server) taskCoverage(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
as := s.m.Assets()
if as == nil {
writeErr(w, 503, "asset store not enabled")
return
}
// Asset coverage off → short-circuits {enabled:false}; the frontend hides the coverage card/progress accordingly.
if !t.CoverageEnabled {
writeJSON(w, 200, &db.Coverage{Enabled: false, ByType: []db.CoverageByType{}})
return
}
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
cov, err := as.TaskCoverageWithSources(taskID)
if err != nil {
writeErr(w, 500, err.Error())
return
}
cov.Enabled = true
writeJSON(w, 200, cov)
}
// taskCoverageGraph returns the force-directed asset coverage graph for a task:
// all in-scope assets (of every type) + the connecting root-domain/company nodes, each carrying tested/in_scope.
func (s *Server) taskCoverageGraph(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
as := s.m.Assets()
if as == nil {
writeErr(w, 503, "asset store not enabled")
return
}
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
g, err := as.BuildCoverageGraph(taskID, t.ExpID)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, g)
}
// taskAssetRefs returns the intents / facts / findings in this task anchored to a
// given asset id — powers the coverage-graph node drawer's "linked intents / linked facts".
func (s *Server) taskAssetRefs(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
assetID, _ := strconv.ParseInt(r.URL.Query().Get("asset_id"), 10, 64)
if assetID <= 0 {
writeErr(w, 400, "asset_id required")
return
}
refs, err := t.Store.AssetRefsWithSources(assetID)
if err != nil {
writeErr(w, 500, err.Error())
return
}
intents := []CoverageAssetRefDTO{}
facts := []CoverageAssetRefDTO{}
findings := []CoverageAssetRefDTO{}
for _, ref := range refs {
dto := coverageAssetRefDTO(ref)
switch ref.Kind {
case "intent":
intents = append(intents, dto)
case "fact":
facts = append(facts, dto)
case "finding":
findings = append(findings, dto)
}
}
writeJSON(w, 200, map[string]any{"intents": intents, "facts": facts, "findings": findings})
}
// taskScopeList returns a task's scope rows (coverage denominator sources).
func (s *Server) taskScopeList(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
as := s.m.Assets()
if as == nil {
writeErr(w, 503, "asset store not enabled")
return
}
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
rows, err := as.ListTaskScopeWithSources(taskID)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"scope": rows})
}
func (s *Server) taskScopeAdd(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
as := s.m.Assets()
if as == nil {
writeErr(w, 503, "asset store not enabled")
return
}
var body struct {
Kind string `json:"kind"`
Value string `json:"value"`
Reason string `json:"reason"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
writeErr(w, 400, "invalid JSON")
return
}
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
ts, err := as.AddAgentScope(taskID, body.Kind, body.Value, body.Reason, "manual")
if err != nil {
writeErr(w, 400, err.Error())
return
}
writeJSON(w, 200, ts)
}
func (s *Server) taskScopeDelete(w http.ResponseWriter, r *http.Request) {
t, ok := s.m.Task(r.PathValue("id"))
if !ok {
writeErr(w, 404, "task not found")
return
}
as := s.m.Assets()
if as == nil {
writeErr(w, 503, "asset store not enabled")
return
}
taskID, _ := strconv.ParseInt(t.ID, 10, 64)
scopeID, err := strconv.ParseInt(r.PathValue("sid"), 10, 64)
if err != nil {
writeErr(w, 400, "invalid scope id")
return
}
deleted, err := as.DeleteTaskScope(taskID, scopeID)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if !deleted {
writeErr(w, 404, "scope row not found")
return
}
writeJSON(w, 200, map[string]bool{"ok": true})
}
func (s *Server) frontier(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeJSON(w, 200, []any{})
return
}
fr, _ := t.Store.Frontier(atoiDefault(r.URL.Query().Get("limit"), 100))
writeJSON(w, 200, taskNodeDTOs(fr))
}
func (s *Server) findings(w http.ResponseWriter, r *http.Request) {
// No task param → the global "Findings" page: read from the standalone findings table (findings survive task deletion).
// With a task param → that task only (task overview/Findings tab), read from exploration_nodes (nodes live as long as the task does).
q := r.URL.Query()
taskParam := q.Get("task")
if taskParam == "" {
// With page/limit → server-side pagination {items,total,...}; without → a bare array (for dashboard aggregation,
// same compat policy as the intents endpoint). Filtering/sorting is pushed down to SQL uniformly.
if q.Get("page") == "" && q.Get("limit") == "" {
fs, _ := s.m.pg.ListFindings(500)
assets := s.resolveFindingAssets(fs)
out := make([]FindingDTO, 0, len(fs))
for _, f := range fs {
out = append(out, findingFromDB(f, assets))
}
writeJSON(w, 200, out)
return
}
page := findingPaginationParam(q.Get("page"), 1, 0)
limit := findingPaginationParam(q.Get("limit"), 20, 200)
fs, total, err := s.m.pg.ListFindingsPage(findingFilterFromQuery(q), page, limit)
if err != nil {
writeErr(w, 500, err.Error())
return
}
assets := s.resolveFindingAssets(fs)
out := make([]FindingDTO, 0, len(fs))
for _, f := range fs {
out = append(out, findingFromDB(f, assets))
}
writeJSON(w, 200, map[string]any{"items": out, "total": total, "page": page, "page_size": limit})
return
}
t := s.m.ResolveTask(taskParam)
if t == nil {
writeJSON(w, 200, []any{})
return
}
f, err := t.Store.ListByKind(db.KindFinding, 200)
if err != nil {
writeErr(w, 500, err.Error())
return
}
tid, _ := strconv.ParseInt(t.ID, 10, 64)
meta, err := s.m.pg.FindingMetaByNodeID(tid)
if err != nil {
log.Printf("[findings] task=%s meta: %v", t.ID, err)
}
var aidSet []int64
for _, m := range meta {
aidSet = append(aidSet, m.AssetIDs...)
}
assets := s.resolveAssetIDs(aidSet)
out := findingDTOsForTask(t, f, meta, assets)
// Related-task findings are a live, read-only view. Provenance metadata lets
// the task UI suppress mutation affordances while retaining stable ids.
sources, err := t.Store.DirectSourceStores()
if err != nil {
writeErr(w, 500, err.Error())
return
}
for _, source := range sources {
nodes, listErr := source.Store.ListByKind(db.KindFinding, 200)
if listErr != nil {
writeErr(w, 500, listErr.Error())
return
}
for _, node := range nodes {
node.SourceTaskID = source.Task.TaskID
node.Inherited = true
}
sourceMeta, metaErr := s.m.pg.FindingMetaByNodeID(source.Task.TaskID)
if metaErr != nil {
log.Printf("[findings] source_task=%d meta: %v", source.Task.TaskID, metaErr)
}
var sourceAssetIDs []int64
for _, item := range sourceMeta {
sourceAssetIDs = append(sourceAssetIDs, item.AssetIDs...)
}
out = append(out, findingDTOsForOwner(
i64s(source.Task.TaskID),
source.Task.Description,
nodes,
sourceMeta,
s.resolveAssetIDs(sourceAssetIDs),
)...)
}
writeJSON(w, 200, out)
}
// normFilter maps the frontend's "all" sentinel (and empty) to "" so the DB layer
// treats it as no filter.
func normFilter(v string) string {
if v == "all" {
return ""
}
return v
}
// resolveFindingAssets loads every asset anchored by the given findings, keyed by
// id, so each finding DTO can render its assets' labels.
func (s *Server) resolveFindingAssets(fs []*db.DBFinding) map[int64]*db.Asset {
var ids []int64
for _, f := range fs {
ids = append(ids, f.AssetIDs...)
}
return s.resolveAssetIDs(ids)
}
// resolveAssetIDs de-dupes ids and loads their asset rows into an id→asset map.
func (s *Server) resolveAssetIDs(ids []int64) map[int64]*db.Asset {
assets := map[int64]*db.Asset{}
seen := map[int64]bool{}
var uniq []int64
for _, id := range ids {
if id > 0 && !seen[id] {
seen[id] = true
uniq = append(uniq, id)
}
}
if len(uniq) == 0 {
return assets
}
list, err := s.m.pg.Assets().GetByIDs(uniq)
if err != nil {
log.Printf("[findings] resolve assets: %v", err)
}
for _, a := range list {
assets[a.ID] = a
}
return assets
}
// findingStats serves the whole-table aggregates (stat cards + vuln-class filter)
// for the paginated Findings page.
func (s *Server) findingStats(w http.ResponseWriter, r *http.Request) {
st, err := s.m.pg.FindingStats()
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, st)
}
// getFinding returns one finding by its standalone-table id (DTO finding_id),
// with anchored assets resolved for display.
func (s *Server) getFinding(w http.ResponseWriter, r *http.Request) {
id := int64(atoiDefault(r.PathValue("id"), 0))
if id <= 0 {
writeErr(w, 400, "bad finding id")
return
}
f, err := s.m.pg.GetFinding(id)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if f == nil {
writeErr(w, 404, "finding not found")
return
}
assets := s.resolveAssetIDs(f.AssetIDs)
dto := findingFromDB(f, assets)
if contextTaskID := strings.TrimSpace(r.URL.Query().Get("context_task")); contextTaskID != "" {
contextTask := s.m.ResolveTask(contextTaskID)
if contextTask == nil {
writeErr(w, 404, "context task not found")
return
}
sourceTaskID, inherited, allowed := findingProvenanceInTask(contextTask, f.TaskID)
if !allowed {
writeErr(w, 404, "finding not available in task context")
return
}
dto.SourceTaskID = sourceTaskID
dto.Inherited = inherited
}
writeJSON(w, 200, dto)
}
// findingsExport exports the Findings page's vulnerabilities.
//
// scope = filtered (keep the page filters) | all (everything) | selected (checked ids)
// format = md-single (one combined .md) | md-zip (one .md per vulnerability, zipped)
// | csv | json
// ids = comma-separated finding ids (required when scope=selected)
// Filter params severity/status/vulnclass/task_id/q/sort match the list endpoint (used when scope=filtered).
func (s *Server) findingsExport(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
scope := q.Get("scope")
format := q.Get("format")
var ids []int64
var filter db.FindingFilter
switch scope {
case "selected":
for _, part := range strings.Split(q.Get("ids"), ",") {
part = strings.TrimSpace(part)
if part == "" {
continue
}
id, err := strconv.ParseInt(part, 10, 64)
if err != nil || id <= 0 {
writeErr(w, 400, "bad finding id: "+part)
return
}
ids = append(ids, id)
}
if len(ids) == 0 {
writeErr(w, 400, "no findings selected")
return
}
case "all":
// empty filter = no conditions added.
case "filtered", "":
filter = findingFilterFromQuery(q)
default:
writeErr(w, 400, "bad scope: "+scope)
return
}
fs, err := s.m.pg.ListFindingsForExport(filter, ids)
if err != nil {
writeErr(w, 500, err.Error())
return
}
stage, err := os.MkdirTemp("", "artex-finding-export-")
if err != nil {
writeErr(w, 500, err.Error())
return
}
defer os.RemoveAll(stage)
if err = s.evidenceStore().StageFindingsExport(r.Context(), fs, stage, format == "md-zip"); err != nil {
evidenceError(w, err)
return
}
now := time.Now()
stamp := now.Format("20060102-150405")
setDownload := func(contentType, filename string) {
w.Header().Set("Content-Type", contentType)
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
}
switch format {
case "md-single":
setDownload("text/markdown; charset=utf-8", "findings-"+stamp+".md")
_, _ = w.Write([]byte(report.FindingsMarkdown(fs, now)))
case "md-zip":
path := filepath.Join(stage, "findings.zip")
if err := buildFindingsEvidenceZip(path, fs, stage, now); err != nil {
writeErr(w, 500, err.Error())
return
}
file, err := os.Open(path)
if err != nil {
writeErr(w, 500, err.Error())
return
}
defer file.Close()
setDownload("application/zip", "findings-"+stamp+".zip")
http.ServeContent(w, r, "findings.zip", now, file)
case "csv":
setDownload("text/csv; charset=utf-8", "findings-"+stamp+".csv")
_, _ = w.Write(report.FindingsCSV(fs))
case "json":
assets := s.resolveFindingAssets(fs)
out := make([]FindingDTO, 0, len(fs))
for _, f := range fs {
for i := range f.TrafficBindings {
f.TrafficBindings[i].Snapshot.ReqHead = ""
f.TrafficBindings[i].Snapshot.RespHead = ""
}
out = append(out, findingFromDB(f, assets))
}
setDownload("application/json; charset=utf-8", "findings-"+stamp+".json")
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
_ = enc.Encode(out)
default:
writeErr(w, 400, "bad format: "+format)
}
}
func findingProvenanceInTask(contextTask *Task, findingTaskID *int64) (sourceTaskID string, inherited, allowed bool) {
if contextTask == nil || findingTaskID == nil {
return "", false, false
}
contextID, err := strconv.ParseInt(contextTask.ID, 10, 64)
if err != nil {
return "", false, false
}
if *findingTaskID == contextID {
return "", false, true
}
for _, sourceID := range contextTask.lifecycleSnapshot().SourceTaskIDs {
if sourceID == *findingTaskID {
return i64s(sourceID), true, true
}
}
return "", false, false
}
// findingLineage returns the exploration sub-DAG from the task root to this
// finding's node — the finding node + all its ancestors + edges among them — so
// the detail page can show "how this finding was reached". Empty {nodes,edges}
// when the finding has no node/task (e.g. the originating task was deleted).
func (s *Server) findingLineage(w http.ResponseWriter, r *http.Request) {
id := int64(atoiDefault(r.PathValue("id"), 0))
if id <= 0 {
writeErr(w, 400, "bad finding id")
return
}
f, err := s.m.pg.GetFinding(id)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if f == nil {
writeErr(w, 404, "finding not found")
return
}
empty := map[string]any{"nodes": []any{}, "edges": []any{}}
if f.NodeID == nil || f.TaskID == nil {
writeJSON(w, 200, empty)
return
}
t := s.m.ResolveTask(i64s(*f.TaskID))
if t == nil {
writeJSON(w, 200, empty)
return
}
nodes, edges, err := t.Store.FindingLineage(*f.NodeID)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"nodes": taskNodeDTOs(nodes), "edges": edgeDTOs(edges)})
}
// patchFinding partially updates a finding: any subset of {status, severity}.
// The id is the standalone findings-table id (DTO finding_id). Severity edits are
// mirrored onto the originating exploration node so the per-task view stays in
// sync. Returns the updated finding DTO.
func (s *Server) patchFinding(w http.ResponseWriter, r *http.Request) {
id := int64(atoiDefault(r.PathValue("id"), 0))
if id <= 0 {
writeErr(w, 400, "bad finding id")
return
}
var body struct {
Status *string `json:"status"`
Severity *string `json:"severity"`
Name *string `json:"name"`
VulnClass *string `json:"vulnclass"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
writeErr(w, 400, "bad json: "+err.Error())
return
}
if body.Status == nil && body.Severity == nil && body.Name == nil && body.VulnClass == nil {
writeErr(w, 400, "nothing to update: provide status/severity/name/vulnclass")
return
}
if body.Status != nil {
if !db.ValidFindingStatus(*body.Status) {
writeErr(w, 400, "bad status: "+*body.Status)
return
}
// Use the notifying variant: the status update and the "status-change push event" land in the same transaction,
// avoiding a window where the status changed but the push event was lost. Event-registration failure does not affect the status update,
// so it is only logged, not reported to the caller.
from, found, notified, err := s.m.pg.SetFindingStatusWithNotify(r.Context(), id, *body.Status)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if !found {
writeErr(w, 404, "finding not found")
return
}
if !notified && from != *body.Status {
log.Printf("[notify] status-change event not registered finding=%d %s→%s (status was updated)", id, from, *body.Status)
}
}
if body.Severity != nil {
if !db.ValidSeverity(*body.Severity) {
writeErr(w, 400, "bad severity: "+*body.Severity)
return
}
n, err := s.m.pg.SetFindingSeverity(id, *body.Severity)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if n == 0 {
writeErr(w, 404, "finding not found")
return
}
}
if body.Name != nil {
n, err := s.m.pg.SetFindingName(id, strings.TrimSpace(*body.Name))
if err != nil {
writeErr(w, 500, err.Error())
return
}
if n == 0 {
writeErr(w, 404, "finding not found")
return
}
}
if body.VulnClass != nil {
n, err := s.m.pg.SetFindingVulnClass(id, strings.TrimSpace(*body.VulnClass))
if err != nil {
writeErr(w, 500, err.Error())
return
}
if n == 0 {
writeErr(w, 404, "finding not found")
return
}
}
f, err := s.m.pg.GetFinding(id)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if f == nil {
writeErr(w, 404, "finding not found")
return
}
writeJSON(w, 200, findingFromDB(f, s.resolveAssetIDs(f.AssetIDs)))
}
// deleteFinding removes a finding (findings row + originating exploration node).
// id is the standalone findings-table id (DTO finding_id).
func (s *Server) deleteFinding(w http.ResponseWriter, r *http.Request) {
id := int64(atoiDefault(r.PathValue("id"), 0))
if id <= 0 {
writeErr(w, 400, "bad finding id")
return
}
n, err := s.m.pg.DeleteFinding(id)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if n == 0 {
writeErr(w, 404, "finding not found")
return
}
writeJSON(w, 200, map[string]any{"deleted": true, "id": id})
}
func (s *Server) intents(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
// Back-compat: bare list shape when the task can't be resolved.
writeJSON(w, 200, []any{})
return
}
q := r.URL.Query()
limit := min(atoiDefault(q.Get("limit"), 300), 500)
// No paging params → preserve the legacy bare-array response so existing callers
// (and the poll) keep working unchanged.
if q.Get("before") == "" && q.Get("page") == "" {
in, err := t.Store.ListByKind(db.KindIntent, limit)
if err != nil {
log.Printf("[intents] task=%s limit=%d: %v", t.ID, limit, err)
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, taskNodeDTOs(in))
return
}
// Paged form: ?before=<id> (or ?page as a marker) → {items, has_more} so the
// worker session list can reach past the old fixed 300 boundary on scroll. It
// also includes immutable intent results from directly related tasks; those
// rows never participate in this task's frontier or claim path.
before := int64(atoiDefault(q.Get("before"), 0))
in, hasMore, err := taskIntentHistoryPage(t.Store, before, limit, false, 0)
if err != nil {
log.Printf("[intents] task=%s before=%d limit=%d: %v", t.ID, before, limit, err)
writeErr(w, 500, err.Error())
return
}
sources, err := t.Store.DirectSourceStores()
if err != nil {
writeErr(w, 500, err.Error())
return
}
for _, source := range sources {
items, sourceMore, sourceErr := taskIntentHistoryPage(source.Store, before, limit, true, source.Task.TaskID)
if sourceErr != nil {
writeErr(w, 500, sourceErr.Error())
return
}
in = append(in, items...)
hasMore = hasMore || sourceMore
}
sort.Slice(in, func(i, j int) bool { return in[i].ID > in[j].ID })
if len(in) > limit {
in = in[:limit]
hasMore = true
}
writeJSON(w, 200, map[string]any{"items": taskNodeDTOs(in), "has_more": hasMore})
}
func inheritedIntentResult(state string) bool {
switch state {
case "done", "blocked", "exhausted", "stopped":
return true
default:
return false
}
}
// inheritedGraphSnapshot exposes immutable source-task history without leaking
// work that is still open or running. Edges are filtered with the nodes so the
// response cannot contain dangling ids that reveal hidden live intent topology.
func inheritedGraphSnapshot(nodes []*db.Node, edges []db.Edge, sourceTaskID int64) ([]*db.Node, []db.Edge) {
visible := make(map[int64]struct{}, len(nodes))
filteredNodes := make([]*db.Node, 0, len(nodes))
for _, node := range nodes {
if node == nil || (node.Kind == db.KindIntent && !inheritedIntentResult(node.State)) {
continue
}
node.SourceTaskID = sourceTaskID
node.Inherited = true
visible[node.ID] = struct{}{}
filteredNodes = append(filteredNodes, node)
}
filteredEdges := make([]db.Edge, 0, len(edges))
for _, edge := range edges {
if _, ok := visible[edge.From]; !ok {
continue
}
if _, ok := visible[edge.To]; !ok {
continue
}
filteredEdges = append(filteredEdges, edge)
}
return filteredNodes, filteredEdges
}
// taskIntentHistoryPage returns one newest-first page for an exploration.
// A source may have many open/running intents, so keep paging until enough
// historical results are collected instead of leaking its frontier into the UI.
func taskIntentHistoryPage(store *db.ExplorationStore, before int64, limit int, inherited bool, sourceTaskID int64) ([]*db.Node, bool, error) {
if !inherited {
return store.ListByKindPage(db.KindIntent, before, limit)
}
batch := max(limit, 300)
cursor := before
out := make([]*db.Node, 0, limit+1)
for {
page, more, err := store.ListByKindPage(db.KindIntent, cursor, batch)
if err != nil {
return nil, false, err
}
for _, node := range page {
if !inheritedIntentResult(node.State) {
continue
}
node.SourceTaskID = sourceTaskID
node.Inherited = true
out = append(out, node)
if len(out) > limit {
return out[:limit], true, nil
}
}
if !more || len(page) == 0 {
return out, false, nil
}
cursor = page[len(page)-1].ID
}
}
// explorationGraph returns the whole exploration chain (task graph) as nodes+edges.
func (s *Server) explorationGraph(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeJSON(w, 200, map[string]any{"nodes": []any{}, "edges": []any{}})
return
}
nodes, err := t.Store.Nodes(2000)
if err != nil {
writeErr(w, 500, err.Error())
return
}
edges, err := t.Store.Edges(5000)
if err != nil {
writeErr(w, 500, err.Error())
return
}
sources, err := t.Store.DirectSourceStores()
if err != nil {
writeErr(w, 500, err.Error())
return
}
for _, source := range sources {
sourceNodes, nodeErr := source.Store.Nodes(2000)
if nodeErr != nil {
writeErr(w, 500, nodeErr.Error())
return
}
sourceEdges, edgeErr := source.Store.Edges(5000)
if edgeErr != nil {
writeErr(w, 500, edgeErr.Error())
return
}
sourceNodes, sourceEdges = inheritedGraphSnapshot(sourceNodes, sourceEdges, source.Task.TaskID)
nodes = append(nodes, sourceNodes...)
edges = append(edges, sourceEdges...)
}
writeJSON(w, 200, map[string]any{"nodes": taskNodeDTOs(nodes), "edges": edgeDTOs(edges)})
}
// explorationNodes serves the broadcast board: this task's own exploration nodes as a
// paged time series (newest first unless ?order=asc), filterable by kind/state
// and a payload substring. Inherited nodes are deliberately out of scope — the
// board reports what this task is doing right now, and paging across the source
// tasks' stores would make the cursor meaningless.
// Response also carries the edges touching the page plus the neighbour nodes
// they point at, so each row can state where it came from and what it produced.
func (s *Server) explorationNodes(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
page := atoiDefault(q.Get("page"), 1)
size := atoiDefault(q.Get("size"), 20)
t := s.m.ResolveTask(q.Get("task"))
if t == nil {
writeJSON(w, 200, map[string]any{
"items": []any{}, "total": 0, "page": page, "size": size,
"edges": []any{}, "refs": map[string]any{},
})
return
}
filter := db.NodeFilter{
Kinds: csvValues(q.Get("kind")),
States: csvValues(q.Get("state")),
Query: q.Get("q"),
Asc: q.Get("order") == "asc",
}
nodes, total, err := t.Store.NodesPage(filter, page, size)
if err != nil {
writeErr(w, 500, err.Error())
return
}
ids := make([]int64, 0, len(nodes))
onPage := make(map[int64]bool, len(nodes))
for _, n := range nodes {
ids = append(ids, n.ID)
onPage[n.ID] = true
}
edges, err := t.Store.EdgesTouching(ids)
if err != nil {
writeErr(w, 500, err.Error())
return
}
neighbourSet := map[int64]bool{}
for _, e := range edges {
if !onPage[e.From] {
neighbourSet[e.From] = true
}
if !onPage[e.To] {
neighbourSet[e.To] = true
}
}
neighbourIDs := make([]int64, 0, len(neighbourSet))
for id := range neighbourSet {
neighbourIDs = append(neighbourIDs, id)
}
neighbours, err := t.Store.NodesByIDs(neighbourIDs)
if err != nil {
writeErr(w, 500, err.Error())
return
}
refs := make(map[string]TaskNodeDTO, len(neighbours))
for _, n := range neighbours {
refs[i64s(n.ID)] = taskNodeDTO(n)
}
writeJSON(w, 200, map[string]any{
"items": taskNodeDTOs(nodes),
"total": total,
"page": page,
"size": size,
"edges": edgeDTOs(edges),
"refs": refs,
"assets": s.nodeAnchoredAssets(t, append(ids, neighbourIDs...)),
})
}
// nodeAnchoredAssets resolves the exploration_anchors of the given nodes into
// display-ready asset labels, keyed by node id. Anchors are provenance decoration
// for the broadcast board — a failure here must not cost the caller its page, so errors are
// logged and degrade to "no assets".
func (s *Server) nodeAnchoredAssets(t *Task, nodeIDs []int64) map[string][]FindingAssetDTO {
out := map[string][]FindingAssetDTO{}
anchors, err := t.Store.NodeAssets(nodeIDs)
if err != nil {
log.Printf("[broadcast] node assets: %v", err)
return out
}
var flat []int64
for _, ids := range anchors {
flat = append(flat, ids...)
}
assets := s.resolveAssetIDs(flat)
for nodeID, ids := range anchors {
if dtos := findingAssetDTOs(ids, assets); len(dtos) > 0 {
out[i64s(nodeID)] = dtos
}
}
return out
}
// csvValues splits a comma-separated query parameter, dropping empty entries.
func csvValues(s string) []string {
out := []string{}
for _, part := range strings.Split(s, ",") {
if v := strings.TrimSpace(part); v != "" {
out = append(out, v)
}
}
return out
}
// activity returns the worker execution step log (incremental via ?since=seq).
func (s *Server) activity(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeJSON(w, 200, map[string]any{"items": []any{}, "cursor": 0})
return
}
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
limit := atoiDefault(r.URL.Query().Get("limit"), 300)
var intentPtr *int64
if iv := r.URL.Query().Get("intent"); iv != "" {
if n, err := strconv.ParseInt(iv, 10, 64); err == nil {
intentPtr = &n
}
}
var (
items []db.Activity
cursor int64
err error
)
if intentPtr != nil {
items, cursor, err = t.Store.ActivityListWithSources(*intentPtr, since, limit)
} else {
items, cursor, err = t.Store.ActivityList(nil, since, limit)
}
if err != nil {
log.Printf("[activity] task=%s since=%d limit=%d intent=%v: %v", t.ID, since, limit, intentPtr, err)
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"items": activityDTOs(items), "cursor": cursor})
}
// parseActivitySession maps a stable session key (main | plan | intent:<ID>) to a
// DB session filter. Goal Agent + Planner both live under worker="planner" (the
// single Plan session); a Worker session is one intent, keyed by its node id.
func parseActivitySession(sess string) (db.ActivitySessionFilter, bool) {
switch {
case sess == "" || sess == "main":
// bare "main" = the current segment; caller resolves MainSeg via the store.
return db.ActivitySessionFilter{Main: true}, true
case strings.HasPrefix(sess, "main:"):
seg, err := strconv.Atoi(strings.TrimPrefix(sess, "main:"))
if err != nil || seg < 0 {
return db.ActivitySessionFilter{}, false
}
return db.ActivitySessionFilter{Main: true, MainSeg: &seg}, true
case sess == "plan":
return db.ActivitySessionFilter{Worker: "planner"}, true
case strings.HasPrefix(sess, "intent:"):
id, err := strconv.ParseInt(strings.TrimPrefix(sess, "intent:"), 10, 64)
if err != nil {
return db.ActivitySessionFilter{}, false
}
return db.ActivitySessionFilter{NodeID: &id}, true
}
return db.ActivitySessionFilter{}, false
}
// activitySessionStore resolves a worker session against the current task and
// its direct sources. Planner/main always stay local; inherited intent sessions
// are immutable history and use the source exploration only for reads.
func activitySessionStore(t *Task, filter db.ActivitySessionFilter) (*db.ExplorationStore, int64, error) {
if filter.NodeID == nil {
return t.Store, 0, nil
}
node, err := t.Store.GetNodeWithSources(*filter.NodeID)
if err != nil {
return nil, 0, err
}
if node == nil || node.Kind != db.KindIntent {
return nil, 0, nil
}
if !node.Inherited {
return t.Store, 0, nil
}
if !inheritedIntentResult(node.State) {
return nil, 0, nil
}
sources, err := t.Store.DirectSourceStores()
if err != nil {
return nil, 0, err
}
for _, source := range sources {
if source.Task.TaskID == node.SourceTaskID {
return source.Store, source.Task.TaskID, nil
}
}
return nil, 0, nil
}
// activityHistory serves one reverse-paginated page of a session's activity history.
// The latest page (no ?before) opens a session; ?before=<id> pulls the older page on
// scroll-up. snapshot_cursor is the TASK-level max id at query time — the client uses
// it to open the single task SSE at since=snapshot_cursor so history (id<=cursor) and
// the live tail (id>cursor) meet with no gap and no overlap.
func (s *Server) activityHistory(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "task not found")
return
}
q := r.URL.Query()
sess := q.Get("session")
filter, ok := parseActivitySession(sess)
if !ok {
writeErr(w, 400, "bad session")
return
}
if filter.Main && filter.MainSeg == nil { // bare "main" → the current segment
seg, err := t.Store.CurrentMainSeg()
if err != nil {
writeErr(w, 500, err.Error())
return
}
filter.MainSeg = &seg
}
before := int64(atoiDefault(q.Get("before"), 0))
limit := min(atoiDefault(q.Get("limit"), 200), 500) // cap so one request can't pull an unbounded slice
store, sourceTaskID, err := activitySessionStore(t, filter)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if store == nil {
writeErr(w, 404, "session not found")
return
}
// The browser tails only this task's broadcaster. Keep the SSE join cursor
// local even when the opened worker transcript comes from a related task.
snapshot, err := t.Store.ActivityMaxID()
if err != nil {
log.Printf("[activity/history] task=%s session=%s snapshot: %v", t.ID, sess, err)
writeErr(w, 500, err.Error())
return
}
var items []db.Activity
var hasMore bool
if sourceTaskID > 0 {
// Source sessions are readable only while the intent remains terminal. The
// DB query also removes model reasoning/accounting rows from inherited data.
items, hasMore, err = store.ActivityPageForTerminalIntent(*filter.NodeID, before, limit)
} else {
items, hasMore, err = store.ActivityPage(filter, before, limit)
}
if err != nil {
log.Printf("[activity/history] task=%s session=%s before=%d limit=%d: %v", t.ID, sess, before, limit, err)
writeErr(w, 500, err.Error())
return
}
if sourceTaskID > 0 {
for i := range items {
items[i].SourceTaskID = sourceTaskID
items[i].Inherited = true
}
}
earliest := before
if len(items) > 0 {
earliest = items[0].ID
}
writeJSON(w, 200, map[string]any{
"items": activityDTOs(items),
"snapshot_cursor": snapshot,
"earliest_cursor": earliest,
"has_more": hasMore,
})
}
// tokenStats returns per-worker token usage (input/output/cache read/write) for a
// task — main agent, planner, and each work#N.
func (s *Server) tokenStats(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeJSON(w, 200, map[string]any{
"workers": []db.TokenUsage{},
"sessions": []db.SessionTokenUsage{},
"total": tokenTotalDTO(db.TokenUsage{}),
})
return
}
stats, err := t.Store.TokenStatsByWorker()
if err != nil {
writeErr(w, 500, err.Error())
return
}
sessions, err := t.Store.TokenStatsBySession()
if err != nil {
writeErr(w, 500, err.Error())
return
}
total, err := t.Store.TokenTotal() // whole-task total (all agents)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"workers": stats, "sessions": sessions, "total": tokenTotalDTO(total)})
}
// tokenDailyStats returns global token consumption aggregated by calendar day
// (UTC) across all tasks for the past ?days=N days (default 30).
func (s *Server) tokenDailyStats(w http.ResponseWriter, r *http.Request) {
days := atoiDefault(r.URL.Query().Get("days"), 30)
if s.m.pg == nil {
writeJSON(w, 200, []any{})
return
}
buckets, err := s.m.pg.TokenDailyAll(days)
if err != nil {
writeErr(w, 500, err.Error())
return
}
if buckets == nil {
buckets = []db.DailyTokenBucket{}
}
writeJSON(w, 200, buckets)
}
// conversationTokens returns per-conversation token summaries so the dashboard can
// merge chat (conversation) usage into its per-profile / daily token stats — which
// otherwise count only task (exploration) usage.
func (s *Server) conversationTokens(w http.ResponseWriter, r *http.Request) {
if s.m.pg == nil {
writeJSON(w, 200, []any{})
return
}
rows, err := s.m.pg.ConversationTokenSummaries()
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, rows)
}
// streamActivity is the live SSE tail: it replays history after ?since=<seq>, then
// pushes each newly appended activity for the task. The seq cursor makes history +
// live join gap-free; on reconnect the client passes its last seq to catch any
// dropped events. Optional ?intent=<id> scopes the stream to one worker session.
// getLogs returns recent backend log lines (Seq > since), newest-last.
func (s *Server) getLogs(w http.ResponseWriter, r *http.Request) {
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
limit := atoiDefault(r.URL.Query().Get("limit"), 500)
lines, cursor := logSink.recent(since, limit)
writeJSON(w, 200, map[string]any{"items": lines, "cursor": cursor})
}
// getLogsHistory returns older log lines from the DB (before a given db_id).
// GET /api/logs/history?before=<db_id>&limit=200
// Returns {items:[LogLine], has_more: bool}.
func (s *Server) getLogsHistory(w http.ResponseWriter, r *http.Request) {
if s.m.pg == nil {
writeJSON(w, 200, map[string]any{"items": []any{}, "has_more": false})
return
}
before := int64(atoiDefault(r.URL.Query().Get("before"), 0))
limit := atoiDefault(r.URL.Query().Get("limit"), 200)
if limit > 500 {
limit = 500
}
// If no before given, return the most recent DB rows (mirrors ring restore).
var (
rows []*db.DBLog
err error
)
if before <= 0 {
rows, err = s.m.pg.RecentLogs(limit)
} else {
rows, err = s.m.pg.ListLogsBefore(before, limit)
}
if err != nil {
writeErr(w, 500, "db: "+err.Error())
return
}
items := make([]LogLine, 0, len(rows))
for _, r := range rows {
items = append(items, LogLine{
DBID: r.ID,
TS: r.CreatedAt.Format(time.RFC3339),
Level: r.Level,
Tag: r.Tag,
Text: r.Text,
})
}
writeJSON(w, 200, map[string]any{"items": items, "has_more": len(rows) == limit})
}
// streamLogs is the live SSE tail of the backend log: replays history after
// ?since=<seq>, then pushes each new line.
func (s *Server) streamLogs(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
writeErr(w, 500, "streaming unsupported")
return
}
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
ch, unsub := logSink.subscribe()
defer unsub()
send := func(l LogLine) {
b, _ := json.Marshal(l)
fmt.Fprintf(w, "data: %s\n\n", b)
flusher.Flush()
}
lines, cursor := logSink.recent(since, 1000)
for _, l := range lines {
send(l)
}
if cursor > since {
since = cursor
}
flusher.Flush()
ctx := r.Context()
ping := time.NewTicker(20 * time.Second)
defer ping.Stop()
for {
select {
case <-ctx.Done():
return
case l, ok := <-ch:
if !ok {
return
}
if l.Seq <= since {
continue
}
since = l.Seq
send(l)
case <-ping.C:
fmt.Fprint(w, ": ping\n\n")
flusher.Flush()
}
}
}
func (s *Server) streamActivity(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "task not found")
return
}
flusher, ok := w.(http.Flusher)
if !ok {
writeErr(w, 500, "streaming unsupported")
return
}
var intentPtr *int64
if iv := r.URL.Query().Get("intent"); iv != "" {
if n, err := strconv.ParseInt(iv, 10, 64); err == nil {
intentPtr = &n
}
}
// Cursor precedence: the browser's automatic reconnect sends Last-Event-ID (the
// last id it received) — trust it over the query so an auto-reconnect resumes
// exactly where it dropped. A fresh/manual connect has no header and passes
// since=<snapshot_cursor> from the history page instead.
since := int64(atoiDefault(r.URL.Query().Get("since"), 0))
if le := r.Header.Get("Last-Event-ID"); le != "" {
if n, err := strconv.ParseInt(le, 10, 64); err == nil {
since = n
}
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering
// Subscribe BEFORE replaying history so events in between aren't lost; dedup the
// overlap by skipping channel events whose id was already replayed.
ch, unsub := s.engine.Broadcaster().Subscribe(t.ID)
defer unsub()
// Emit a standard SSE id: line so the browser echoes it as Last-Event-ID on
// auto-reconnect (see cursor precedence above).
sendSSE := func(a db.Activity) {
b, _ := json.Marshal(activityDTO(a))
fmt.Fprintf(w, "id: %d\ndata: %s\n\n", a.ID, b)
flusher.Flush()
}
// Compensate the DB backlog after `since` in batches until caught up. This is the
// gap between the history snapshot and the live tail — NOT the first-page history
// (that's the /activity/history endpoint). A long task can have far more than one
// batch, so loop instead of a single fixed read; on a query error log it and close
// so the client reconnects and retries from its last id (broadcast is lossy — the
// DB is the source of truth). The Broadcaster keeps buffering live events meanwhile;
// the id<=since skip below drops any that this replay already covered.
const replayBatch = 500
for {
items, cursor, err := t.Store.ActivityList(intentPtr, since, replayBatch)
if err != nil {
log.Printf("[activity/stream] task=%s replay since=%d: %v", t.ID, since, err)
return
}
for _, a := range items {
sendSSE(a)
}
if cursor > since {
since = cursor
}
if len(items) < replayBatch {
break
}
}
flusher.Flush()
ctx := r.Context()
ping := time.NewTicker(20 * time.Second)
defer ping.Stop()
for {
select {
case <-ctx.Done():
return
case a, ok := <-ch:
if !ok {
return
}
if a.ID <= since {
continue // already replayed
}
if intentPtr != nil && (a.NodeID == nil || *a.NodeID != *intentPtr) {
continue // scoped session: only this intent's steps
}
since = a.ID
sendSSE(a)
case <-ping.C:
fmt.Fprint(w, ": ping\n\n")
flusher.Flush()
}
}
}
// activityDetail lazily returns the full detail blob for one step.
func (s *Server) activityDetail(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "no task")
return
}
seq, _ := strconv.ParseInt(r.PathValue("seq"), 10, 64)
d, err := taskActivityDetail(t, seq)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"detail": d})
}
// taskActivityDetail keeps the legacy local-task behavior (including thinking
// rows), while inherited details are restricted to terminal worker intents.
// This prevents a guessed global activity id from exposing source planner/main
// or in-flight transcripts.
func taskActivityDetail(t *Task, seq int64) (string, error) {
return t.Store.ActivityDetailWithSources(seq)
}
func (s *Server) getTraffic(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeJSON(w, 200, map[string]any{"enabled": false, "exchanges": []any{}})
return
}
q := r.URL.Query()
page := atoiDefault(q.Get("page"), 0)
size := atoiDefault(q.Get("size"), 100)
ex, matched, _ := tr.Page(traffic.PageQuery{
Host: q.Get("host"),
Method: q.Get("method"),
Query: q.Get("q"),
Body: q.Get("body"),
Path: q.Get("path"),
Status: q.Get("status"),
RespMin: int64(atoiDefault(q.Get("resp_min"), -1)),
RespMax: int64(atoiDefault(q.Get("resp_max"), -1)),
Sort: q.Get("sort"),
Order: q.Get("order"),
}, page, size)
count, _ := tr.Count() // global total, for the stat card
writeJSON(w, 200, map[string]any{
"enabled": s.m.TrafficEnabled(), // reflect the capture toggle
"proxy": s.m.ProxyAddr(),
"count": count, // total recorded (unfiltered)
"total": matched, // rows matching the current filter (for pagination)
"page": page,
"size": size,
"exchanges": trafficDTOs(ex),
})
}
// getTrafficHosts returns distinct recorded hosts with counts, for the page's
// target picker (pick a host → filter the list, then delete it).
func (s *Server) getTrafficHosts(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeJSON(w, 200, map[string]any{"hosts": []any{}})
return
}
hosts, err := tr.Hosts()
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"hosts": hosts})
}
// deleteTraffic removes recorded traffic for every host containing the query's
// host substring (the page's host filter is substring-based, so what you
// filtered is what gets deleted): index rows + each host's file tree, then
// garbage-collects blobs no remaining exchange references. Empty host → 400.
// Returns the number of exchanges deleted.
func (s *Server) deleteTraffic(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeErr(w, 404, "traffic disabled")
return
}
host := strings.TrimSpace(r.URL.Query().Get("host"))
if host == "" {
writeErr(w, 400, "missing host")
return
}
n, err := tr.DeleteHost(host)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"deleted": n})
}
// deleteTrafficHosts removes traffic for a set of EXACT hosts (JSON body
// {"hosts": [...]}) — the batch path for the page's multi-select delete. Exact
// match, so picking "api.example.com" never sweeps "api.example.com.cn".
// Returns the number of exchanges deleted.
func (s *Server) deleteTrafficHosts(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeErr(w, 404, "traffic disabled")
return
}
var req struct {
Hosts []string `json:"hosts"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, "invalid body")
return
}
hosts := make([]string, 0, len(req.Hosts))
for _, h := range req.Hosts {
if h = strings.TrimSpace(h); h != "" {
hosts = append(hosts, h)
}
}
if len(hosts) == 0 {
writeErr(w, 400, "missing hosts")
return
}
n, err := tr.DeleteHostsExact(hosts)
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"deleted": n})
}
// deleteAllTraffic purges every recorded exchange, then compacts the index so
// the space is actually returned to the filesystem — an emptied index is the one
// moment a full rewrite is cheap. Evidence already bound to a finding lives in
// the evidence store and is deliberately left alone. Returns the number of
// exchanges deleted and the bytes of index reclaimed.
func (s *Server) deleteAllTraffic(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeErr(w, 404, "traffic disabled")
return
}
n, reclaimed, err := tr.DeleteAll()
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"deleted": n, "reclaimed": reclaimed})
}
// getTrafficExchange returns the full raw request/response of one exchange,
// read on demand from the traffic tree (bodies are not in the paged list).
func (s *Server) getTrafficExchange(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeErr(w, 404, "traffic disabled")
return
}
id := r.URL.Query().Get("id")
if id == "" {
writeErr(w, 400, "missing id")
return
}
req, resp, err := tr.Get(id)
if err != nil {
writeErr(w, 404, err.Error())
return
}
writeJSON(w, 200, map[string]any{"req": req, "resp": resp})
}
// getTrafficBlob streams one oversized body by its sha256. Bodies past the inline
// threshold are not carried by the exchange endpoint — it returns a preview plus
// an "@blob sha256:<hash>" pointer — so this is how the UI fetches them whole.
// Streamed rather than buffered: these are the bodies too large to hold in memory.
func (s *Server) getTrafficBlob(w http.ResponseWriter, r *http.Request) {
tr := s.m.Traffic()
if tr == nil {
writeErr(w, 404, "traffic disabled")
return
}
hash := strings.TrimSpace(r.URL.Query().Get("hash"))
if hash == "" {
writeErr(w, 400, "missing hash")
return
}
f, size, err := tr.Blob(hash)
if err != nil {
writeErr(w, 404, err.Error())
return
}
defer f.Close()
w.Header().Set("Content-Type", "application/octet-stream")
w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", hash+".bin"))
if _, err := io.Copy(w, f); err != nil {
log.Printf("[traffic] blob %s download interrupted: %v", hash, err)
}
}
// getSettings returns the runtime app settings the UI toggles. The Brave API key
// is returned as a boolean presence flag (brave_key_set), never the value itself,
// so the UI can show "configured" without echoing the secret back.
func (s *Server) getSettings(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, s.settingsPayload())
}
func (s *Server) settingsPayload() map[string]any {
on, backend, braveKey, tavilyKey, proxy := s.m.WebSearch()
pyStored, _, _ := s.m.pg.GetSetting(settingPythonInterp)
concOn, concLimit := s.m.ConcurrencyLimit()
if concLimit == 0 {
concLimit = defaultConcurrencyLimit // when off, still echo a sane default to the UI
}
return map[string]any{
"traffic_capture": s.m.TrafficEnabled(),
"agent_traffic_binding": s.m.pg.GetBool(settingAgentTrafficBinding, false),
"llm_record": s.m.LLMRecordEnabled(),
"web_search_enabled": on,
"web_search_backend": backend,
"brave_key_set": strings.TrimSpace(braveKey) != "",
"tavily_key_set": strings.TrimSpace(tavilyKey) != "",
"web_search_proxy": proxy, // dedicated egress proxy (http/https/socks5), empty = direct
"global_proxy": s.m.GlobalProxy(), // global egress proxy (http/https/socks5), all target traffic goes through it, empty = direct
"python_interpreter": strings.TrimSpace(pyStored), // user/auto-set value (empty = use runtime detection)
"workers": s.m.Workers(), // number of concurrent worker agents (default 3); applies to tasks started afterwards
"task_concurrency_enabled": concOn, // task concurrency cap switch (default off)
"task_concurrency_limit": concLimit, // max simultaneously running tasks (default 5 when enabled)
// LLM failover pool. Default off; when on, agents on the global active profile automatically
// switch to the next profile when the current one is unavailable. bind_fallback only matters when the pool is on (default off).
"llm_pool_enabled": s.m.LLMPoolEnabled(),
"llm_pool_bind_fallback": s.m.LLMPoolBindFallback(),
// operation-constraint injection scope (default both on): splices this task's allow/deny constraints into the matching agents' system prompts.
"constraints_inject_planner": s.constraintInjectPlanner(),
"constraints_inject_worker": s.constraintInjectWorker(),
// Experimental: noa model-driven context compaction (default off). When on, noa takes over context
// compaction for the platform's four agent kinds, replacing built-in compaction; read once per run, applies to runs started afterwards.
"noa_compaction": s.m.NoaCompactionEnabled(),
// Vulnerability IM push global settings. Channels are standalone resources managed via /api/notify/*;
// only the three items that "apply to all channels" live here.
"notify_enabled": s.m.pg.GetBool(settingNotifyEnabled, true),
"notify_public_base_url": notifyPublicBaseURL(s.m.pg),
"notify_digest_interval_min": notifyDigestIntervalMin(s.m.pg),
}
}
// notifyPublicBaseURL reads the external address used in push backlinks.
func notifyPublicBaseURL(pg *db.DB) string {
v, _, _ := pg.GetSetting(settingNotifyPublicBaseURL)
return v
}
// notifyDigestIntervalMin reads the digest interval (minutes), falling back to the default when invalid or unset.
// Echo the default instead of an empty string so the UI can fill the effective value into the input.
func notifyDigestIntervalMin(pg *db.DB) int {
v, ok, _ := pg.GetSetting(settingNotifyDigestMinutes)
if !ok {
return notifyDefaultDigestMinutes
}
n, err := strconv.Atoi(strings.TrimSpace(v))
if err != nil || n <= 0 {
return notifyDefaultDigestMinutes
}
return n
}
// pgDetectPython re-runs interpreter detection, stores + returns it.
func (s *Server) pgDetectPython(w http.ResponseWriter, r *http.Request) {
p := detectPython()
if p == "" {
writeErr(w, 404, "no python detected (neither python3 nor python is on PATH)")
return
}
if err := s.m.pg.SetSetting(settingPythonInterp, p); err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"python_interpreter": p})
}
// putSettings applies a settings change. Toggling traffic_capture rebuilds the
// agents (applyLLM) so the new proxy/traffic-tools/prompt state takes hold — when
// off, agents get no proxy config, no traffic tools, and no proxy prompt content.
func (s *Server) putSettings(w http.ResponseWriter, r *http.Request) {
var req struct {
TrafficCapture *bool `json:"traffic_capture"`
AgentTrafficBinding *bool `json:"agent_traffic_binding"`
LLMRecord *bool `json:"llm_record"` // LLM recording switch (default off); takes effect immediately, no agent rebuild needed
// Web search. WebSearchEnabled/Backend toggle the tool + backend; BraveKey/TavilyKey
// are optional — omit (null) to leave a stored key untouched, send "" to clear.
WebSearchEnabled *bool `json:"web_search_enabled"`
WebSearchBackend *string `json:"web_search_backend"`
BraveKey *string `json:"brave_search_api_key"`
TavilyKey *string `json:"tavily_search_api_key"`
WebSearchProxy *string `json:"web_search_proxy"` // dedicated egress proxy (http/https/socks5); null = keep, "" = clear
GlobalProxy *string `json:"global_proxy"` // global egress proxy (http/https/socks5); null = keep, "" = clear (direct)
PythonInterp *string `json:"python_interpreter"` // python interpreter path for the custom script tool
Workers *int `json:"workers"` // number of concurrent worker agents (>0); applies to tasks started afterwards
// task concurrency cap: max simultaneously "running" tasks. Off = unlimited; on, new tasks beyond the cap queue and start automatically as slots free up.
ConcurrencyEnabled *bool `json:"task_concurrency_enabled"`
ConcurrencyLimit *int `json:"task_concurrency_limit"`
// LLM failover pool switch + the "fall back to the pool chain even when a bound profile fails" switch. Both need the
// provider chain rebuilt to take effect, via the changed → applyLLM path below.
LLMPoolEnabled *bool `json:"llm_pool_enabled"`
LLMPoolBindFallback *bool `json:"llm_pool_bind_fallback"`
// operation-constraint injection scope switches (default both on); take effect immediately (planner/worker read them every round), no agent rebuild needed.
ConstraintsInjectPlanner *bool `json:"constraints_inject_planner"`
ConstraintsInjectWorker *bool `json:"constraints_inject_worker"`
// Experimental: noa context compaction switch (default off); read per run, applies to runs started afterwards, no agent rebuild needed.
NoaCompaction *bool `json:"noa_compaction"`
// Vulnerability IM push global settings. The delivery engine re-reads all three every round, so changes
// take effect immediately with no agent rebuild or restart.
NotifyEnabled *bool `json:"notify_enabled"`
NotifyBaseURL *string `json:"notify_public_base_url"`
NotifyDigestMins *int `json:"notify_digest_interval_min"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
if req.ConstraintsInjectPlanner != nil {
if err := s.m.pg.SetBool(settingConstraintsInjectPlanner, *req.ConstraintsInjectPlanner); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.ConstraintsInjectWorker != nil {
if err := s.m.pg.SetBool(settingConstraintsInjectWorker, *req.ConstraintsInjectWorker); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.NoaCompaction != nil {
// The resolver is read per run; toggling takes effect immediately for runs started afterwards, no applyLLM rebuild needed.
if err := s.m.SetNoaCompaction(*req.NoaCompaction); err != nil {
writeErr(w, 500, err.Error())
return
}
}
// Push global settings: the delivery engine re-reads them every round, so they take effect immediately without a restart.
if req.NotifyEnabled != nil {
if err := s.m.pg.SetBool(settingNotifyEnabled, *req.NotifyEnabled); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.NotifyBaseURL != nil {
// Trim trailing slashes uniformly: backlinks are joined with fmt.Sprintf("%s/function/..."),
// and a trailing slash would produce double-slash paths like "//function/...".
base := trimTrailingSlash(strings.TrimSpace(*req.NotifyBaseURL))
if base != "" && !strings.HasPrefix(base, "http://") && !strings.HasPrefix(base, "https://") {
writeErr(w, 400, "backlink base URL must start with http:// or https://")
return
}
if err := s.m.pg.SetSetting(settingNotifyPublicBaseURL, base); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.NotifyDigestMins != nil {
// 1-minute floor: any shorter period equals realtime push, in which case the channel should just be set to realtime mode.
if *req.NotifyDigestMins < 1 || *req.NotifyDigestMins > 24*60 {
writeErr(w, 400, "digest interval must be between 1 and 1440 minutes")
return
}
if err := s.m.pg.SetSetting(settingNotifyDigestMinutes, strconv.Itoa(*req.NotifyDigestMins)); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.Workers != nil {
if err := s.m.SetWorkers(*req.Workers); err != nil {
writeErr(w, 400, err.Error())
return
}
}
if req.ConcurrencyEnabled != nil || req.ConcurrencyLimit != nil {
// Partial PUT: fields not supplied fall back to current values, so changing one doesn't reset the other.
curOn, curLimit := s.m.ConcurrencyLimit()
if curLimit == 0 {
curLimit = defaultConcurrencyLimit
}
on, limit := curOn, curLimit
if req.ConcurrencyEnabled != nil {
on = *req.ConcurrencyEnabled
}
if req.ConcurrencyLimit != nil {
limit = *req.ConcurrencyLimit
}
if err := s.m.SetConcurrency(on, limit); err != nil {
writeErr(w, 400, err.Error())
return
}
// Reconcile immediately: turning off releases all queued tasks, raising the cap backfills starts — no need to wait for the next tick.
go s.reconcileConcurrency()
}
if req.PythonInterp != nil {
if err := s.m.pg.SetSetting(settingPythonInterp, strings.TrimSpace(*req.PythonInterp)); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.LLMRecord != nil {
// The recorder reads this flag on every call; toggling takes effect immediately, no applyLLM rebuild needed.
if err := s.m.SetLLMRecordEnabled(*req.LLMRecord); err != nil {
writeErr(w, 500, err.Error())
return
}
}
changed := false
if req.LLMPoolEnabled != nil {
if err := s.m.SetLLMPoolEnabled(*req.LLMPoolEnabled); err != nil {
writeErr(w, 500, err.Error())
return
}
changed = true // the provider chain itself changes shape → rebuild
}
if req.LLMPoolBindFallback != nil {
if err := s.m.SetLLMPoolBindFallback(*req.LLMPoolBindFallback); err != nil {
writeErr(w, 500, err.Error())
return
}
changed = true
}
if req.LLMPoolEnabled != nil || req.LLMPoolBindFallback != nil {
// Pinned tasks' planner/worker and per-profile chat agents hold providers
// built under the OLD switch state — drop them so they pick up the new one.
s.invalidateProfileAgents()
}
if req.AgentTrafficBinding != nil {
if err := s.m.pg.SetBool(settingAgentTrafficBinding, *req.AgentTrafficBinding); err != nil {
writeErr(w, 500, err.Error())
return
}
}
if req.TrafficCapture != nil {
if err := s.m.SetTrafficEnabled(*req.TrafficCapture); err != nil {
writeErr(w, 500, err.Error())
return
}
changed = true
}
if req.GlobalProxy != nil {
// Validation failure (bad scheme/host) is a client error, not a 500.
if err := s.m.SetGlobalProxy(*req.GlobalProxy); err != nil {
writeErr(w, 400, err.Error())
return
}
changed = true // capture-off egress is baked into agents at build time → rebuild
}
if req.WebSearchEnabled != nil || req.WebSearchBackend != nil || req.BraveKey != nil || req.TavilyKey != nil || req.WebSearchProxy != nil {
// Fill unspecified fields from current state so a partial PUT doesn't reset them.
on, backend, _, _, _ := s.m.WebSearch()
if req.WebSearchEnabled != nil {
on = *req.WebSearchEnabled
}
if req.WebSearchBackend != nil {
backend = *req.WebSearchBackend
}
if err := s.m.SetWebSearch(on, backend, req.BraveKey, req.TavilyKey, req.WebSearchProxy); err != nil {
writeErr(w, 500, err.Error())
return
}
changed = true
}
if changed {
// rebuild agents so the new proxy/tools/prompt/web-search take hold (only if LLM configured).
s.cfgMu.Lock()
cfg, on := s.llmCfg, s.llmOn
s.cfgMu.Unlock()
if on {
if err := s.applyLLM(cfg); err != nil {
writeErr(w, 500, err.Error())
return
}
}
}
writeJSON(w, 200, s.settingsPayload())
}
// testWebSearch runs a real "test" search with the given (or currently saved)
// backend/proxy/key to verify the config can actually reach a search backend —
// mirroring testLLM. Backend/proxy come from the request (so the form's unsaved
// edits are tested); empty API keys fall back to stored values so the user need
// not retype them. Always 200 with {ok, error?, count?, backend?}.
func (s *Server) testWebSearch(w http.ResponseWriter, r *http.Request) {
var req struct {
Backend string `json:"web_search_backend"`
Proxy string `json:"web_search_proxy"`
BraveKey string `json:"brave_search_api_key"`
TavilyKey string `json:"tavily_search_api_key"`
}
// Empty body is fine — fall back entirely to the saved config below.
if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err != io.EOF {
writeErr(w, 400, err.Error())
return
}
_, backend, storedBraveKey, storedTavilyKey, _ := s.m.WebSearch()
if strings.TrimSpace(req.Backend) != "" {
backend = req.Backend
}
// Proxy is taken from the form as-is (empty = direct), so testing reflects exactly
// what's shown — including an intentional "clear proxy to test direct" before saving.
proxy := strings.TrimSpace(req.Proxy)
// API keys are secrets the form omits when already saved, so fall back to stored.
braveKey := storedBraveKey
if strings.TrimSpace(req.BraveKey) != "" {
braveKey = req.BraveKey
}
tavilyKey := storedTavilyKey
if strings.TrimSpace(req.TavilyKey) != "" {
tavilyKey = req.TavilyKey
}
cfg := actool.WebSearchConfig{Backend: backend, BraveAPIKey: braveKey, TavilyAPIKey: tavilyKey, Proxy: proxy}
// Hard cap so a slow/blocked proxy can't hang the request.
wall := 30 * time.Second
// deepseek's credentials are not in the form; they come from the currently active LLM profile. Each search also runs one
// model inference, so the generic 30s cap is tight — relaxed just for it. We don't pre-judge whether the profile works: this test
// exists precisely for the user to confirm, and a real failure's error below says more than any pre-check.
probeQuery := "test"
if strings.TrimSpace(backend) == deepSeekWebSearchBackend {
cfg.DeepSeekBaseURL, cfg.DeepSeekAPIKey, cfg.DeepSeekModel = s.m.deepSeekSearchCreds()
wall = 120 * time.Second
// The search term is decided by the model on DeepSeek's side; "test" is too vague and would make it skip the search and answer directly.
probeQuery = "DeepSeek company official website"
}
ctx, cancel := context.WithTimeout(r.Context(), wall)
defer cancel()
results, err := actool.WebSearchProbe(ctx, cfg, probeQuery, 3)
if err != nil {
writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error(), "backend": backend})
return
}
if len(results) == 0 {
writeJSON(w, 200, map[string]any{"ok": false, "error": "search returned 0 results (possibly rate-limited or the proxy is unreachable)", "backend": backend})
return
}
writeJSON(w, 200, map[string]any{"ok": true, "count": len(results), "backend": backend})
}
// mainSessions lists the task's main-agent conversation segments (newest-first) and
// the current one. The frontend renders these as switchable sessions under Main Agent.
func (s *Server) mainSessions(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "task not found")
return
}
list, err := t.Store.ListMainSessions()
if err != nil {
writeErr(w, 500, err.Error())
return
}
current := 0
if len(list) > 0 {
current = list[0].Seq // newest-first
}
writeJSON(w, 200, map[string]any{"sessions": list, "current": current})
}
// newMainSession starts a fresh main-agent conversation segment. Only the segment
// counter advances — the task's exploration graph, assets and goal are untouched, so
// the main agent continues over the same task with a clean transcript/context.
func (s *Server) newMainSession(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "task not found")
return
}
if s.engine.IsDeleting(t.ID) {
writeErr(w, 409, "task is being deleted; cannot create a new session")
return
}
m, err := t.Store.NewMainSession()
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"seq": m.Seq, "created_at": rfc3339(m.CreatedAt), "current": m.Seq})
}
func (s *Server) chat(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "no active task")
return
}
if s.engine.IsDeleting(t.ID) {
writeErr(w, 409, "task is being deleted; cannot send a new message")
return
}
// Note: a paused task does not block main-agent conversations. The main-agent orchestration session is independent of the
// planner/worker pause; chat continues while paused (pausing only aborts the turn in flight — see control()).
var req struct {
Message string `json:"message"`
Attachments []chatAttachment `json:"attachments,omitempty"` // files uploaded via method 1 (paths relative to the task work dir)
Seg *int `json:"seg,omitempty"` // target main-session segment; omitted = newest segment
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error())
return
}
agentMessage, ok := s.prepareChatMentionMessage(w, req.Message)
if !ok {
return
}
// Admission is serialized with deletion's chat cancellation. Mark every chat
// turn busy before its first activity/file write, including rule-mode turns.
// If deletion wins the race, the second barrier check rejects this request.
s.chatMu.Lock()
if s.engine.IsDeleting(t.ID) {
s.chatMu.Unlock()
writeErr(w, 409, "task is being deleted; cannot send a new message")
return
}
if s.chatBusy[t.ID] {
s.chatMu.Unlock()
writeErr(w, 409, "the main agent is still processing the previous message, please wait")
return
}
ctx, cancel := context.WithCancelCause(s.ctx)
ctx = intercept.WithReviewContext(ctx, "", intercept.ReviewBackground{Source: intercept.BackgroundUserMessage, Text: req.Message})
s.chatBusy[t.ID] = true
s.chatCancel[t.ID] = cancel
s.chatMu.Unlock()
// The turn belongs to whichever main-agent segment the user is chatting in (any
// segment is interactive, like the top-level chat conversations). Stamp every
// mainagent row with it so this turn's transcript + activity land in that segment.
// Missing seg (older clients) falls back to the newest segment.
mainSeg := 0
if req.Seg != nil && *req.Seg >= 0 {
mainSeg = *req.Seg
} else {
mainSeg, _ = t.Store.CurrentMainSeg()
}
segPtr := &mainSeg
// Persist + broadcast the human turn so the main-agent orchestration session survives page
// reloads and updates live: the conversation lives in the activity stream as
// worker="mainagent" (the per-task activity table, replayed via SSE). With
// attachments, the activity's Detail carries {text, attachments} so the transcript
// renders attachment cards.
humanTurn := userActivityWithAttachments("mainagent", req.Message, req.Attachments)
humanTurn.MainSeg = segPtr
s.engine.emitActivity(t, humanTurn)
var ma *agent.MainAgent
if s.taskRuntimeAvailable(t, "mainagent") {
ma = s.agentsForTask(t).main
}
if ma != nil {
// Run the agent on a per-turn cancellable ctx (NOT r.Context()): the turn can
// take minutes (multi-tool loop), and binding it to the request lifecycle meant
// a page reload / proxy timeout cancelled it mid-run ("context canceled"). The
// steps + final answer stream back live via SSE (worker="mainagent"), so the
// handler returns immediately and the browser never needs to hold the request.
go func() {
defer func() {
s.finishTaskChat(t.ID, cancel)
}()
// emit every step (thinking/tool_use/tool_result/text/result) so the main
// agent session shows its work live, like worker/planner. The final answer
// is the captured "result" step — no separate reply emit (would duplicate).
emit := func(rec db.Activity) {
rec.MainSeg = segPtr
s.engine.emitActivity(t, rec)
}
maTaskID, _ := strconv.ParseInt(t.ID, 10, 64)
resume := func() { s.reviveTask(t) } // set_goals added a goal → pulls the task back to running
// Splice the list of [absolute paths] of uploaded attachments into the message sent to the agent; it uses them to open files via Read/Bash.
// taskDir = the agent's working directory (CWD), consistent with chatUpload's target dir and ensureRunDir.
taskDir := filepath.Join(s.m.dir, "tasks", t.ID)
agentMsg := composeAgentMessage(agentMessage, req.Attachments, taskDir)
s.engine.BeginLLMCall(t.ID)
_, err := ma.Chat(ctx, maTaskID, mainSeg, s.m.Assets(), t.Store, t.Goal, agentMsg, emit, t.Notify, resume, t.NotifyGoal, t.NotifyHint)
s.engine.EndLLMCall(t.ID)
if err != nil && ctx.Err() == nil {
s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", IsError: true, Summary: "(main agent error: " + err.Error() + ")", MainSeg: segPtr})
}
}()
writeJSON(w, 202, map[string]any{"status": "accepted", "mode": "llm"})
return
}
reply := s.fallbackChat(t, req.Message)
s.engine.emitActivity(t, db.Activity{Worker: "mainagent", Kind: "text", Summary: reply, MainSeg: segPtr})
s.finishTaskChat(t.ID, cancel)
writeJSON(w, 200, map[string]any{"reply": reply, "mode": "rule"})
}
func (s *Server) cancelTaskChat(taskID string, cause error) bool {
s.chatMu.Lock()
cancel := s.chatCancel[taskID]
if cancel != nil {
cancel(cause)
}
s.chatMu.Unlock()
return cancel != nil
}
func (s *Server) finishTaskChat(taskID string, cancel context.CancelCauseFunc) {
cancel(agent.AbortChatTurnFinished)
s.chatMu.Lock()
delete(s.chatBusy, taskID)
delete(s.chatCancel, taskID)
s.chatMu.Unlock()
}
// taskChatStatus reports the authoritative state of the task's main-agent turn.
// Activity timestamps are not a reliable proxy because a tool or LLM call may run
// for minutes without emitting an intermediate frame.
func (s *Server) taskChatStatus(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.PathValue("id"))
if t == nil {
writeErr(w, http.StatusNotFound, "task not found")
return
}
s.chatMu.Lock()
running := s.chatBusy[t.ID]
s.chatMu.Unlock()
writeJSON(w, http.StatusOK, map[string]bool{"running": running})
}
// stopChat aborts the in-flight main-agent turn for a task (manual stop button).
// Mirrors pgStopConversation: cancels only the current turn; planner/workers are
// unaffected and continue running. The user can send a new message immediately.
func (s *Server) stopChat(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.PathValue("id"))
if t == nil {
writeErr(w, 404, "task not found")
return
}
if !s.cancelTaskChat(t.ID, agent.AbortChatStoppedByUser) {
writeJSON(w, 200, map[string]any{"status": "idle"})
return
}
writeJSON(w, 200, map[string]any{"status": "stopping"})
}
// fallbackChat is the no-LLM human-steering handler: simple commands + situation summary.
func (s *Server) fallbackChat(t *Task, msg string) string {
m := strings.TrimSpace(msg)
lower := strings.ToLower(m)
switch {
case strings.HasPrefix(m, "intent") || strings.HasPrefix(lower, "intent"):
text := strings.TrimSpace(strings.TrimPrefix(strings.TrimPrefix(m, "intent"), "intent"))
_, _ = t.Store.AddIntent(map[string]any{"summary": text}, 9, nil, "human")
return "Injected one high-priority intent: " + text
case strings.HasPrefix(m, "hint") || strings.HasPrefix(lower, "hint"):
text := strings.TrimSpace(strings.TrimPrefix(strings.TrimPrefix(m, "hint"), "hint"))
_, _ = t.Store.AddNode(db.KindHint, map[string]any{"text": text}, 0, "active", "human", nil)
return "Hint recorded; the planner will read it next time: " + text
default:
assetCounts, _ := s.m.Assets().CountsByType()
assets := 0
for _, c := range assetCounts {
assets += c
}
fnd, _ := t.Store.ListByKind(db.KindFinding, 1000)
fr, _ := t.Store.Frontier(1000)
return fmt.Sprintf("(rule mode, no LLM configured) Current situation: assets %d, open intents %d, confirmed findings %d.\nAvailable commands: \"intent ...\" injects an intent, \"hint ...\" leaves the planner a hint.", assets, len(fr), len(fnd))
}
}
func (s *Server) getReport(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeErr(w, 404, "no active task")
return
}
findings, _ := t.Store.ListByKind(db.KindFinding, 1000) // pure vulnerabilities (facts are a separate KindFact and never enter the report)
counts := map[string]int{}
for _, ty := range []string{"root_domain", "ip", "subdomain", "app", "service", "endpoint"} {
ns, _ := s.m.Assets().QueryByType(ty, 100000, 0)
if len(ns) > 0 {
counts[ty] = len(ns)
}
}
md := report.Markdown(report.Input{
Title: t.Description, Goal: t.Goal, GeneratedAt: time.Now(),
AssetCounts: counts, Findings: findings,
})
w.Header().Set("Content-Type", "text/markdown; charset=utf-8")
w.WriteHeader(200)
_, _ = w.Write([]byte(md))
}
func (s *Server) getAudit(w http.ResponseWriter, r *http.Request) {
t := s.m.ResolveTask(r.URL.Query().Get("task"))
if t == nil {
writeJSON(w, 200, []any{})
return
}
writeJSON(w, 200, map[string]any{"entries": t.Guard.Audit(), "attributions": t.Guard.Attributions()})
}
// gc is a no-op stub (GC not yet implemented in the new asset store).
func (s *Server) gc(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"removed": 0})
}
// --- utils ---
func cors(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
if r.Method == http.MethodOptions {
w.WriteHeader(204)
return
}
next.ServeHTTP(w, r)
})
}
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(v)
}
func writeErr(w http.ResponseWriter, code int, msg string) {
writeJSON(w, code, map[string]any{"error": msg})
}
func atoiDefault(s string, d int) int {
if n, err := strconv.Atoi(s); err == nil {
return n
}
return d
}