mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-14 20:20:54 +08:00
Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b269d3e590 | ||
|
|
908832ae1b | ||
|
|
9f03b8c16a | ||
|
|
8d841a6011 | ||
|
|
35bd98561c | ||
|
|
d0ffd96960 | ||
|
|
ac8491d666 | ||
|
|
f57cf659b5 |
@@ -1192,6 +1192,28 @@ WHERE exploration_id=$1 AND kind='goal' AND state='open')`, s.expID).Scan(&exist
|
||||
return exists, err
|
||||
}
|
||||
|
||||
// OpenGoalFingerprint returns a stable fingerprint of the open-goal set: the
|
||||
// sorted goal ids joined by commas, "" when none. Planner quiet-mode (#79) uses
|
||||
// it to tell "the goal set changed since the last planning round" apart from
|
||||
// "nothing the plan depends on changed".
|
||||
func (s *ExplorationStore) OpenGoalFingerprint() (string, error) {
|
||||
rows, err := s.db.Query(`SELECT id FROM exploration_nodes
|
||||
WHERE exploration_id=$1 AND kind='goal' AND state='open' ORDER BY id`, s.expID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer rows.Close()
|
||||
var ids []string
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return "", err
|
||||
}
|
||||
ids = append(ids, strconv.FormatInt(id, 10))
|
||||
}
|
||||
return strings.Join(ids, ","), rows.Err()
|
||||
}
|
||||
|
||||
// ClaimIntent atomically moves an open intent to running. Returns true if claimed.
|
||||
func (s *ExplorationStore) ClaimIntent(id int64, owner string) (bool, error) {
|
||||
res, err := s.db.Exec(`UPDATE exploration_nodes SET state='running', owner=$1
|
||||
|
||||
@@ -526,6 +526,7 @@ CREATE TABLE IF NOT EXISTS tasks (
|
||||
parent_ref TEXT,
|
||||
timeout_seconds INTEGER NOT NULL DEFAULT 0,
|
||||
plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300,
|
||||
plan_quiet_mode BOOLEAN NOT NULL DEFAULT true,
|
||||
coverage_enabled BOOLEAN NOT NULL DEFAULT true,
|
||||
pinned_at TIMESTAMPTZ,
|
||||
first_run_at TIMESTAMPTZ,
|
||||
@@ -543,6 +544,7 @@ CREATE TRIGGER trg_tasks_upd BEFORE UPDATE ON tasks
|
||||
FOR EACH ROW EXECUTE FUNCTION set_updated_at();
|
||||
-- planner heartbeat trigger interval (seconds); backfills old databases. Default 300s (5min). See docs/planner-trigger-impl-plan.md
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_heartbeat_seconds INTEGER NOT NULL DEFAULT 300;
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS plan_quiet_mode BOOLEAN NOT NULL DEFAULT true;
|
||||
-- Concurrency-limit hold state; backfills old databases. true=queued due to the concurrency limit, auto-starts when a slot frees up.
|
||||
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS queued BOOLEAN NOT NULL DEFAULT false;
|
||||
-- Asset coverage feature switch; backfills old databases. true (default)=compute/show test coverage, auto-accumulate test scope,
|
||||
|
||||
+20
-7
@@ -46,6 +46,9 @@ type Task struct {
|
||||
// last plan round ended / the task started, with no trigger in between → trigger a round.
|
||||
// Floor = default = 300 (5min); anything lower is raised to 300 (normalized in CreateTask). See docs/planner-trigger-impl-plan.md
|
||||
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"`
|
||||
// PlanQuietMode gates the planner quiet-mode (#86): edge-triggered rounds
|
||||
// are skipped while work is in flight and the open-goal set is unchanged.
|
||||
PlanQuietMode bool `json:"plan_quiet_mode"`
|
||||
// CoverageEnabled is the master switch of the "asset coverage" feature (default true). When false:
|
||||
// do not compute/show test coverage, do not auto-accumulate task_scope (source=auto), do not open
|
||||
// add_task_scope/list_untested_assets to the agent, and do not inject the coverage block into the
|
||||
@@ -154,6 +157,8 @@ type TaskCreateOptions struct {
|
||||
LLMProfileIDs []int64
|
||||
TimeoutSeconds int
|
||||
PlanHeartbeatSeconds int
|
||||
// PlanQuietMode is the per-task quiet-mode switch; nil = default on (#86).
|
||||
PlanQuietMode *bool
|
||||
// CoverageEnabled is the "asset coverage" feature switch; nil=default on (true), letting
|
||||
// creation paths that do not care about the switch (orchestration spawn, old API) keep the
|
||||
// original behavior. Only web task creation can explicitly pass false to turn it off.
|
||||
@@ -219,6 +224,10 @@ VALUES ($1, 'fact', $2, 0, 'origin', 'system')`, expID, string(originPayload));
|
||||
id := opts.LLMProfileIDs[0]
|
||||
active = &id
|
||||
}
|
||||
quietMode := true
|
||||
if opts.PlanQuietMode != nil {
|
||||
quietMode = *opts.PlanQuietMode
|
||||
}
|
||||
t := &Task{
|
||||
Name: opts.Name, CategoryID: opts.CategoryID, CategoryName: categoryName,
|
||||
Description: description, Goal: goal, ExplorationID: expID,
|
||||
@@ -227,12 +236,13 @@ VALUES ($1, 'fact', $2, 0, 'origin', 'system')`, expID, string(originPayload));
|
||||
SourceTaskIDs: append([]int64(nil), opts.SourceTaskIDs...),
|
||||
CompanyIDs: append([]int64(nil), opts.CompanyIDs...),
|
||||
TimeoutSeconds: opts.TimeoutSeconds, PlanHeartbeatSeconds: opts.PlanHeartbeatSeconds,
|
||||
PlanQuietMode: quietMode,
|
||||
CoverageEnabled: coverageEnabled,
|
||||
}
|
||||
if err := tx.QueryRow(`
|
||||
INSERT INTO tasks(name, category_id, description, goal, exploration_id, llm_profile_id, active_llm_profile_id, timeout_seconds, plan_heartbeat_seconds, coverage_enabled)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$6,$7,$8,$9)
|
||||
RETURNING id, status, paused, created_at`, opts.Name, opts.CategoryID, description, goal, expID, active, opts.TimeoutSeconds, opts.PlanHeartbeatSeconds, coverageEnabled).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil {
|
||||
INSERT INTO tasks(name, category_id, description, goal, exploration_id, llm_profile_id, active_llm_profile_id, timeout_seconds, plan_heartbeat_seconds, plan_quiet_mode, coverage_enabled)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$6,$7,$8,$9,$10)
|
||||
RETURNING id, status, paused, created_at`, opts.Name, opts.CategoryID, description, goal, expID, active, opts.TimeoutSeconds, opts.PlanHeartbeatSeconds, quietMode, coverageEnabled).Scan(&t.ID, &t.Status, &t.Paused, &t.CreatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := insertTaskRelations(tx, t.ID, opts.SourceTaskIDs); err != nil {
|
||||
@@ -348,11 +358,11 @@ func insertTaskLLMProfiles(tx *sql.Tx, taskID int64, profileIDs []int64) error {
|
||||
|
||||
const taskCols = `id, COALESCE(name,''), category_id,
|
||||
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
|
||||
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), first_run_at, deadline_at`
|
||||
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(plan_quiet_mode,true), COALESCE(coverage_enabled,true), first_run_at, deadline_at`
|
||||
|
||||
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
|
||||
var t Task
|
||||
if err := sc.Scan(&t.ID, &t.Name, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.CoverageEnabled, &t.FirstRunAt, &t.DeadlineAt); err != nil {
|
||||
if err := sc.Scan(&t.ID, &t.Name, &t.CategoryID, &t.CategoryName, &t.Description, &t.Goal, &t.ExplorationID, &t.Status, &t.Paused, &t.Queued, &t.QueuedAt, &t.QueueMode, &t.LLMProfileID, &t.ActiveLLMProfileID, &t.ParentRef, &t.PinnedAt, &t.CreatedAt, &t.CompletedAt, &t.TimeoutSeconds, &t.PlanHeartbeatSeconds, &t.PlanQuietMode, &t.CoverageEnabled, &t.FirstRunAt, &t.DeadlineAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
t.Pinned = t.PinnedAt != nil
|
||||
@@ -396,6 +406,8 @@ ORDER BY (pinned_at IS NOT NULL) DESC, pinned_at DESC NULLS LAST, id DESC`)
|
||||
type TaskPatch struct {
|
||||
Name *string
|
||||
Pinned *bool
|
||||
// PlanQuietMode flips the per-task quiet-mode switch (#86); nil = leave unchanged.
|
||||
PlanQuietMode *bool
|
||||
}
|
||||
|
||||
// UpdateTask applies a partial task name/pin mutation and returns the updated row.
|
||||
@@ -407,9 +419,10 @@ func (d *DB) UpdateTask(id int64, patch TaskPatch) (*Task, error) {
|
||||
WHEN $4::boolean IS NULL THEN pinned_at
|
||||
WHEN $4::boolean THEN COALESCE(pinned_at, now())
|
||||
ELSE NULL
|
||||
END
|
||||
END,
|
||||
plan_quiet_mode = CASE WHEN $5::boolean THEN $6 ELSE plan_quiet_mode END
|
||||
WHERE id=$1 AND deleted_at IS NULL
|
||||
RETURNING `+taskCols, id, patch.Name != nil, patch.Name, patch.Pinned))
|
||||
RETURNING `+taskCols, id, patch.Name != nil, patch.Name, patch.Pinned, patch.PlanQuietMode != nil, patch.PlanQuietMode))
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
@@ -61,6 +61,7 @@ type TaskDTO struct {
|
||||
ArchiveBlockedBy string `json:"archive_blocked_by_task_id,omitempty"`
|
||||
CompanyIDs []int64 `json:"company_ids"`
|
||||
CoverageEnabled bool `json:"coverage_enabled"` // asset coverage feature switch (fixed at creation)
|
||||
PlanQuietMode bool `json:"plan_quiet_mode"` // planner quiet-mode switch (#86), editable mid-run
|
||||
}
|
||||
|
||||
// FindingSeverityDTO is the per-severity finding count shown in the task list (critical/high/medium/low).
|
||||
@@ -134,6 +135,7 @@ func taskDTO(t *Task, status string) TaskDTO {
|
||||
SourceTaskIDs: sourceIDs,
|
||||
CompanyIDs: lifecycle.CompanyIDs,
|
||||
CoverageEnabled: t.CoverageEnabled,
|
||||
PlanQuietMode: t.PlanQuietMode,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -99,6 +99,10 @@ type Engine struct {
|
||||
started sync.Map // taskID -> bool, so Run is idempotent per task
|
||||
lastAct sync.Map // taskID -> int64 unix, last planner/worker activity (heartbeat)
|
||||
llmCalls sync.Map // taskID -> *int64, actual planner/worker/main-agent LLM calls
|
||||
// quiet-mode (#79): open-goal fingerprint recorded at the end of the last
|
||||
// completed planning round, per task. Edge-triggered rounds are skipped
|
||||
// while it is unchanged and work is still in flight.
|
||||
goalFP sync.Map // taskID -> string (sorted open-goal ids joined)
|
||||
paused sync.Map // taskID -> bool, user-paused (planner + workers idle but loops alive)
|
||||
deleting sync.Map // taskID -> bool, delete barrier (no new task-owned writes)
|
||||
dropCnt sync.Map // taskID -> *int64, running count of dropped (unpersistable) activity records
|
||||
@@ -797,6 +801,22 @@ func resetPlannerTimer(timer *time.Timer, d time.Duration) {
|
||||
timer.Reset(d)
|
||||
}
|
||||
|
||||
// quietSkipEdge reports whether an edge-triggered planning round should be
|
||||
// dropped (#79): the open-goal set is unchanged since the last completed round
|
||||
// AND active work exists (open or running intents). Fails open (false) on any
|
||||
// store error: when in doubt, plan, preserving the pre-quiet-mode behavior.
|
||||
func (e *Engine) quietSkipEdge(t *Task) bool {
|
||||
fp, err := t.Store.OpenGoalFingerprint()
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
if last, ok := e.goalFP.Load(t.ID); !ok || last != fp {
|
||||
return false
|
||||
}
|
||||
active, err := t.Store.HasActiveIntent()
|
||||
return err == nil && active
|
||||
}
|
||||
|
||||
func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
interval := plannerHeartbeatInterval(t)
|
||||
// Arming the heartbeat timer at loop entry = counting from the task start: even a
|
||||
@@ -856,6 +876,17 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
}
|
||||
return // the goalless branch never enters planner.Plan
|
||||
}
|
||||
// quiet-mode (#79): an edge round is redundant while work is in flight
|
||||
// and the open-goal set is unchanged since the last completed round —
|
||||
// there is no new decision for the plan to make. Drain the triggers so
|
||||
// they do not accumulate and let the heartbeat supervise; the next edge
|
||||
// after a goal opens or closes, or once the frontier drains and all
|
||||
// workers are idle, plans for real.
|
||||
if src == "edge" && t.PlanQuietMode && e.quietSkipEdge(t) {
|
||||
t.drainTriggers()
|
||||
log.Printf("[planner] task %s quiet: work in flight, goals unchanged, edge round skipped", t.ID)
|
||||
return
|
||||
}
|
||||
if !e.beginTaskOperation(t.ID) {
|
||||
return
|
||||
}
|
||||
@@ -895,6 +926,12 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
default:
|
||||
log.Printf("[planner] task %s planning finished", t.ID)
|
||||
}
|
||||
// Record the post-round open-goal fingerprint for quiet-mode (#79):
|
||||
// goals opened or closed by this round change it, so a later edge can
|
||||
// tell "nothing changed" apart from "the plan input changed".
|
||||
if fp, err := t.Store.OpenGoalFingerprint(); err == nil {
|
||||
e.goalFP.Store(t.ID, fp)
|
||||
}
|
||||
e.touch(t.ID)
|
||||
}
|
||||
|
||||
|
||||
@@ -6,17 +6,32 @@ import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/Autumn-27/artex/db"
|
||||
)
|
||||
|
||||
func TestInheritedActivityDetailAndRelationDeletion(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
// #78: t.TempDir() cleanup raced with background writers that outlive
|
||||
// m.Close (TempDir RemoveAll: directory not empty, under CI load). Use a
|
||||
// manual dir with a tolerated cleanup, and close the manager via t.Cleanup
|
||||
// registered FIRST so it runs LAST (LIFO): the DeleteTask cleanups below
|
||||
// then execute while the database is still open instead of racing a closed
|
||||
// one ("sql: database is closed" teardown noise).
|
||||
dir, err := os.MkdirTemp("", "inherited-activity-*")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, err := NewManager(dir, "")
|
||||
if err != nil {
|
||||
_ = os.RemoveAll(dir)
|
||||
t.Skipf("postgres unavailable (%v) - skipping", err)
|
||||
}
|
||||
defer m.Close()
|
||||
t.Cleanup(func() {
|
||||
_ = m.Close()
|
||||
_ = os.RemoveAll(dir) // tolerated: a late background writer may still hold it
|
||||
})
|
||||
|
||||
source, err := m.CreateTaskWithOptions("detail source", "source goal", db.TaskCreateOptions{})
|
||||
if err != nil {
|
||||
|
||||
@@ -57,6 +57,7 @@ type Task struct {
|
||||
// Task-level timeout (see docs/task-timeout-and-settle-design.md). DeadlineAt/FirstRunAt are unix seconds, 0 = unset/not yet run.
|
||||
TimeoutSeconds int `json:"timeout_seconds"`
|
||||
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` // planner heartbeat trigger interval (seconds)
|
||||
PlanQuietMode bool `json:"plan_quiet_mode"` // planner quiet-mode switch (#86), editable mid-run
|
||||
CoverageEnabled bool `json:"coverage_enabled"` // asset coverage feature switch (fixed at creation, on by default)
|
||||
FirstRunAt int64 `json:"first_run_at,omitempty"`
|
||||
DeadlineAt int64 `json:"deadline_at,omitempty"`
|
||||
@@ -844,6 +845,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
|
||||
SourceTaskIDs: append([]int64(nil), pt.SourceTaskIDs...),
|
||||
CompanyIDs: append([]int64(nil), pt.CompanyIDs...),
|
||||
TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds,
|
||||
PlanQuietMode: pt.PlanQuietMode,
|
||||
CoverageEnabled: pt.CoverageEnabled,
|
||||
FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt),
|
||||
Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1),
|
||||
@@ -873,6 +875,9 @@ func (m *Manager) UpdateTaskMetadata(taskID string, patch pgdb.TaskPatch) (*Task
|
||||
state.Name = updated.Name
|
||||
state.PinnedAt = unixOrZero(updated.PinnedAt)
|
||||
})
|
||||
// Live switch (#86): runRound reads the flag per round, so the next
|
||||
// planning round (edge or heartbeat) already respects the new value.
|
||||
task.PlanQuietMode = updated.PlanQuietMode
|
||||
return task, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/Autumn-27/artex/db"
|
||||
)
|
||||
|
||||
// Planner quiet-mode (#79): an edge round is dropped only when BOTH hold — the
|
||||
// open-goal fingerprint is unchanged since the last completed round AND active
|
||||
// work exists (open or running intents). Every other combination plans.
|
||||
func TestPlannerQuietSkipEdge(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = m.Close() })
|
||||
if m.pg == nil {
|
||||
t.Skip("no postgres for this run")
|
||||
}
|
||||
|
||||
created, err := m.CreateTaskWithOptions("quiet-mode probe", "probe goal", db.TaskCreateOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tt, ok := m.Task(created.ID)
|
||||
if !ok {
|
||||
t.Fatal("task not found after creation")
|
||||
}
|
||||
store := tt.Store
|
||||
e := NewEngine(m)
|
||||
|
||||
gid, err := store.AddGoal(map[string]any{"summary": "g1"}, "seed")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
iid, err := store.AddIntent(map[string]any{"summary": "i1"}, 0, nil, "seed")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fp, err := store.OpenGoalFingerprint()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if fp == "" {
|
||||
t.Fatal("fingerprint must not be empty with an open goal")
|
||||
}
|
||||
e.goalFP.Store(tt.ID, fp)
|
||||
if !e.quietSkipEdge(tt) {
|
||||
t.Fatal("unchanged goals + active intent must skip the edge round")
|
||||
}
|
||||
|
||||
// a goal closes -> the fingerprint changes -> plan
|
||||
if err := store.SetNodeState(gid, "met"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if e.quietSkipEdge(tt) {
|
||||
t.Fatal("a closed goal must allow planning")
|
||||
}
|
||||
|
||||
// goal back open, intent drained -> no active work -> plan
|
||||
if err := store.SetNodeState(gid, "open"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fp2, err := store.OpenGoalFingerprint()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
e.goalFP.Store(tt.ID, fp2)
|
||||
if err := store.SetIntentState(iid, "done"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if e.quietSkipEdge(tt) {
|
||||
t.Fatal("no active intent must allow planning")
|
||||
}
|
||||
}
|
||||
|
||||
// The quiet-mode switch is editable mid-run (#86): PATCHing plan_quiet_mode
|
||||
// updates both the stored row and the live Task the engine reads per round.
|
||||
func TestPlannerQuietModeToggleMidRun(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = m.Close() })
|
||||
if m.pg == nil {
|
||||
t.Skip("no postgres for this run")
|
||||
}
|
||||
|
||||
created, err := m.CreateTaskWithOptions("quiet-toggle probe", "goal", db.TaskCreateOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tt, ok := m.Task(created.ID)
|
||||
if !ok {
|
||||
t.Fatal("task not found")
|
||||
}
|
||||
if !tt.PlanQuietMode {
|
||||
t.Fatal("quiet mode must default on")
|
||||
}
|
||||
|
||||
off := false
|
||||
updated, err := m.UpdateTaskMetadata(created.ID, db.TaskPatch{PlanQuietMode: &off})
|
||||
if err != nil || updated == nil {
|
||||
t.Fatalf("UpdateTaskMetadata: %v %v", updated, err)
|
||||
}
|
||||
live, ok := m.Task(created.ID)
|
||||
if !ok || live.PlanQuietMode {
|
||||
t.Fatal("mid-run toggle must turn the live flag off")
|
||||
}
|
||||
stored, err := m.pg.GetTask(mustTaskID(t, created.ID))
|
||||
if err != nil || stored == nil || stored.PlanQuietMode {
|
||||
t.Fatalf("stored row must reflect the toggle: %v %v", stored, err)
|
||||
}
|
||||
|
||||
on := true
|
||||
if _, err := m.UpdateTaskMetadata(created.ID, db.TaskPatch{PlanQuietMode: &on}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
live, _ = m.Task(created.ID)
|
||||
if !live.PlanQuietMode {
|
||||
t.Fatal("toggle back on failed")
|
||||
}
|
||||
}
|
||||
+35
-19
@@ -1044,34 +1044,44 @@ func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) {
|
||||
active = t.ID
|
||||
}
|
||||
list := s.m.List()
|
||||
metrics, _ := s.m.PG().TaskListMetricsAll()
|
||||
metrics, err := s.m.PG().TaskListMetricsAll()
|
||||
if err != nil {
|
||||
log.Printf("[tasks] list metrics query failed — metrics zeroed: %v", err)
|
||||
}
|
||||
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,
|
||||
}
|
||||
s.applyTaskMetrics(&dto, t, metrics[t.ExpID])
|
||||
dtos = append(dtos, dto)
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"tasks": dtos, "active": active})
|
||||
}
|
||||
|
||||
// applyTaskMetrics fills the aggregate DTO fields (tokens, goal progress,
|
||||
// running intents, findings, last activity) from the list-metrics query so the
|
||||
// detail and list endpoints report identical numbers. A zero metric leaves the
|
||||
// DTO zeroed, which is why callers log the query error instead of assuming it.
|
||||
func (s *Server) applyTaskMetrics(dto *TaskDTO, t *Task, metric db.TaskListMetrics) {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) resolvedTaskStatus(t *Task) string {
|
||||
if t == nil {
|
||||
return "created"
|
||||
@@ -1494,6 +1504,7 @@ type createTaskReq struct {
|
||||
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
|
||||
PlanQuietMode *bool `json:"plan_quiet_mode,omitempty"` // planner quiet-mode switch (#86); omitted = default on; false = edge rounds always plan
|
||||
SeedFirstIntent *bool `json:"seed_first_intent,omitempty"` // issue one seed intent at creation (content = description + goal) so a worker can start right away without waiting for the first planner round; omitted/null = default off, the standard plan-then-execute flow. Pass true explicitly to enable (for CTF tasks often solved in one work, this skips the pre-start planner round).
|
||||
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).
|
||||
@@ -1562,7 +1573,7 @@ func (s *Server) createTask(w http.ResponseWriter, r *http.Request) {
|
||||
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,
|
||||
TimeoutSeconds: req.TimeoutSeconds, PlanHeartbeatSeconds: req.PlanHeartbeatSeconds, PlanQuietMode: req.PlanQuietMode,
|
||||
CoverageEnabled: req.CoverageEnabled,
|
||||
InterceptRules: interceptRules,
|
||||
})
|
||||
@@ -1797,6 +1808,11 @@ func (s *Server) getTask(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
dto := taskDTO(t, s.resolvedTaskStatus(t))
|
||||
metrics, err := s.m.PG().TaskListMetricsAll()
|
||||
if err != nil {
|
||||
log.Printf("[tasks] detail metrics query failed for %s — metrics zeroed: %v", t.ID, err)
|
||||
}
|
||||
s.applyTaskMetrics(&dto, t, metrics[t.ExpID])
|
||||
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
|
||||
applyTaskArchiveBlocker(&dto, archiveBlockers)
|
||||
writeJSON(w, 200, dto)
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTaskDetailReturnsListMetrics(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) — skipping", err)
|
||||
}
|
||||
defer m.Close()
|
||||
task, err := m.CreateTask("detail metrics", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
taskID, _ := strconv.ParseInt(task.ID, 10, 64)
|
||||
defer func() { _ = m.pg.DeleteTask(taskID) }()
|
||||
|
||||
// Seed the aggregates the metrics query reads: one result-activity with
|
||||
// tokens, one non-result row that must be excluded, and two goals (1 met).
|
||||
if _, err := m.pg.Exec(`
|
||||
INSERT INTO activity(
|
||||
exploration_id, kind, input_tokens, output_tokens,
|
||||
cache_read_tokens, cache_write_tokens, created_at
|
||||
) VALUES ($1,'result',11,7,3,2,now()), ($1,'tool_result',999,999,999,999,now())`,
|
||||
task.ExpID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := m.pg.Exec(`
|
||||
INSERT INTO exploration_nodes(exploration_id, kind, payload, state)
|
||||
VALUES ($1,'goal','{}','met'), ($1,'goal','{}','open')`, task.ExpID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
s := New(context.Background(), m, t.TempDir(), t.TempDir(), t.TempDir())
|
||||
token, err := signJWT(s.jwtKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var probe struct {
|
||||
Tokens struct {
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
CacheReadTokens int `json:"cache_read_tokens"`
|
||||
CacheWriteTokens int `json:"cache_write_tokens"`
|
||||
} `json:"tokens"`
|
||||
GoalsMet int `json:"goals_met"`
|
||||
GoalsTotal int `json:"goals_total"`
|
||||
}
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/tasks/"+task.ID, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
rec := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("detail status=%d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &probe); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if probe.Tokens.InputTokens != 11 || probe.Tokens.OutputTokens != 7 ||
|
||||
probe.Tokens.CacheReadTokens != 3 || probe.Tokens.CacheWriteTokens != 2 {
|
||||
t.Fatalf("detail tokens not filled from metrics: %+v", probe.Tokens)
|
||||
}
|
||||
if probe.GoalsMet != 1 || probe.GoalsTotal != 2 {
|
||||
t.Fatalf("detail goal progress not filled from metrics: %d/%d", probe.GoalsMet, probe.GoalsTotal)
|
||||
}
|
||||
|
||||
// The list endpoint must report the same numbers for the same task.
|
||||
listReq := httptest.NewRequest(http.MethodGet, "/api/tasks", nil)
|
||||
listReq.Header.Set("Authorization", "Bearer "+token)
|
||||
listRec := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(listRec, listReq)
|
||||
if listRec.Code != http.StatusOK {
|
||||
t.Fatalf("list status=%d body=%s", listRec.Code, listRec.Body.String())
|
||||
}
|
||||
var listed struct {
|
||||
Tasks []struct {
|
||||
ID string `json:"id"`
|
||||
Tokens struct {
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
} `json:"tokens"`
|
||||
GoalsMet int `json:"goals_met"`
|
||||
GoalsTotal int `json:"goals_total"`
|
||||
} `json:"tasks"`
|
||||
}
|
||||
if err := json.Unmarshal(listRec.Body.Bytes(), &listed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
found := false
|
||||
for _, lt := range listed.Tasks {
|
||||
if lt.ID != task.ID {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if lt.Tokens.InputTokens != probe.Tokens.InputTokens ||
|
||||
lt.Tokens.OutputTokens != probe.Tokens.OutputTokens ||
|
||||
lt.GoalsMet != probe.GoalsMet || lt.GoalsTotal != probe.GoalsTotal {
|
||||
t.Fatalf("list and detail metrics disagree: list=%+v/%d detail=%+v/%d",
|
||||
lt.Tokens, lt.GoalsMet, probe.Tokens, probe.GoalsTotal)
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("task %s missing from list response", task.ID)
|
||||
}
|
||||
}
|
||||
@@ -25,8 +25,9 @@ func (s *Server) updateTaskMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
r.Body = http.MaxBytesReader(w, r.Body, maxTaskMetadataRequestSize)
|
||||
var request struct {
|
||||
Name *string `json:"name"`
|
||||
Pinned *bool `json:"pinned"`
|
||||
Name *string `json:"name"`
|
||||
Pinned *bool `json:"pinned"`
|
||||
PlanQuietMode *bool `json:"plan_quiet_mode"`
|
||||
}
|
||||
if err := decode(r, &request); err != nil {
|
||||
var tooLarge *http.MaxBytesError
|
||||
@@ -37,8 +38,8 @@ func (s *Server) updateTaskMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
return
|
||||
}
|
||||
if request.Name == nil && request.Pinned == nil {
|
||||
writeErr(w, http.StatusBadRequest, "at least one of name or pinned must be provided")
|
||||
if request.Name == nil && request.Pinned == nil && request.PlanQuietMode == nil {
|
||||
writeErr(w, http.StatusBadRequest, "at least one of name, pinned or plan_quiet_mode must be provided")
|
||||
return
|
||||
}
|
||||
if request.Name != nil {
|
||||
@@ -53,7 +54,7 @@ func (s *Server) updateTaskMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
request.Name = &name
|
||||
}
|
||||
task, err := s.m.UpdateTaskMetadata(taskID, db.TaskPatch{Name: request.Name, Pinned: request.Pinned})
|
||||
task, err := s.m.UpdateTaskMetadata(taskID, db.TaskPatch{Name: request.Name, Pinned: request.Pinned, PlanQuietMode: request.PlanQuietMode})
|
||||
if err != nil {
|
||||
writeErr(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user