Merge pull request #60 from RuoJi6/fix/task-runtime-exclude-pause

fix: 运行时长排除暂停时间
This commit is contained in:
RuoJi6 authored and GitHub committed 2026-10-27 00:44:03 -07:00
commit cb0a69ef60
16 files changed
+587 -88

No files matched your search

+1 -1
View File
@@ -1272,7 +1272,7 @@ func (d *DB) TokenTotalsAll() (map[int64]TokenUsage, error) {
// LastActivityAll returns the unix time of the most recent activity per // LastActivityAll returns the unix time of the most recent activity per
// exploration (exploration_id → max created_at epoch), one query for all tasks. // exploration (exploration_id → max created_at epoch), one query for all tasks.
// Persisted (unlike Engine.LastActivity's in-memory map), so it survives restarts // Persisted (unlike Engine.LastActivity's in-memory map), so it survives restarts
// and gives终态任务 a stable "ran until" time for computing run duration. // and gives终态任务 a stable last-activity time for history views.
func (d *DB) LastActivityAll() (map[int64]int64, error) { func (d *DB) LastActivityAll() (map[int64]int64, error) {
rows, err := d.Query(`SELECT exploration_id, EXTRACT(EPOCH FROM MAX(created_at))::bigint FROM activity GROUP BY exploration_id`) rows, err := d.Query(`SELECT exploration_id, EXTRACT(EPOCH FROM MAX(created_at))::bigint FROM activity GROUP BY exploration_id`)
if err != nil { if err != nil {
+4
View File
@@ -528,6 +528,8 @@ CREATE TABLE IF NOT EXISTS tasks (
pinned_at TIMESTAMPTZ, pinned_at TIMESTAMPTZ,
first_run_at TIMESTAMPTZ, first_run_at TIMESTAMPTZ,
deadline_at TIMESTAMPTZ, deadline_at TIMESTAMPTZ,
active_run_seconds BIGINT NOT NULL DEFAULT 0,
run_segment_started_at TIMESTAMPTZ,
archived_at TIMESTAMPTZ, archived_at TIMESTAMPTZ,
deleted_at TIMESTAMPTZ, deleted_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ, completed_at TIMESTAMPTZ,
@@ -562,6 +564,8 @@ ALTER TABLE tasks ADD COLUMN IF NOT EXISTS pinned_at TIMESTAMPTZ;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active_llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active_llm_profile_id BIGINT REFERENCES llm_profiles(id) ON DELETE SET NULL;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS llm_chain_revision BIGINT NOT NULL DEFAULT 0; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS llm_chain_revision BIGINT NOT NULL DEFAULT 0;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS archived_at TIMESTAMPTZ; ALTER TABLE tasks ADD COLUMN IF NOT EXISTS archived_at TIMESTAMPTZ;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active_run_seconds BIGINT NOT NULL DEFAULT 0;
ALTER TABLE tasks ADD COLUMN IF NOT EXISTS run_segment_started_at TIMESTAMPTZ;
CREATE INDEX IF NOT EXISTS idx_tasks_category ON tasks(category_id, created_at DESC) CREATE INDEX IF NOT EXISTS idx_tasks_category ON tasks(category_id, created_at DESC)
WHERE deleted_at IS NULL AND category_id IS NOT NULL; WHERE deleted_at IS NULL AND category_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_tasks_pinned ON tasks(pinned_at DESC) CREATE INDEX IF NOT EXISTS idx_tasks_pinned ON tasks(pinned_at DESC)
+172
View File
@@ -0,0 +1,172 @@
package db
import (
"database/sql"
"testing"
)
func TestTaskRunSegmentLifecycle(t *testing.T) {
d, err := Open(testDSN(t))
if err != nil {
t.Skipf("postgres unavailable (%v) — skipping", err)
}
defer d.Close()
task, err := d.CreateTask("runtime lifecycle", "runtime lifecycle test", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
active, started, err := d.StartTaskRunSegment(task.ID)
if err != nil {
t.Fatal(err)
}
if started == nil {
t.Fatal("StartTaskRunSegment did not open a segment")
}
if active != 0 {
t.Fatalf("new task active seconds=%d, want 0", active)
}
if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
active, started, err = d.FinalizeTaskRunSegment(task.ID)
if err != nil {
t.Fatal(err)
}
if active <= 0 || started != nil {
t.Fatalf("first finalize active=%d started=%v, want positive and NULL", active, started)
}
firstActive := active
// Pause folds the open segment exactly once and closes it. Repeated
// finalization after wall-clock advancement remains a no-op.
if _, _, err := d.StartTaskRunSegment(task.ID); err != nil {
t.Fatal(err)
}
if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
active, started, err = d.SetPausedRuntime(task.ID, true)
if err != nil {
t.Fatal(err)
}
if active <= firstActive || started != nil {
t.Fatalf("pause active=%d started=%v, want increase and NULL", active, started)
}
pausedActive := active
if active, started, err = d.FinalizeTaskRunSegment(task.ID); err != nil {
t.Fatal(err)
} else if active != pausedActive || started != nil {
t.Fatalf("repeated finalize after pause active=%d started=%v, want %d and NULL", active, started, pausedActive)
}
// Resuming opens a fresh segment, whose elapsed time is added on top of the
// paused total rather than replacing it.
if _, _, err := d.SetPausedRuntime(task.ID, false); err != nil {
t.Fatal(err)
}
if _, started, err = d.StartTaskRunSegment(task.ID); err != nil {
t.Fatal(err)
} else if started == nil {
t.Fatal("resume did not open a segment")
}
if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
active, started, err = d.FinalizeTaskRunSegment(task.ID)
if err != nil {
t.Fatal(err)
}
if active <= pausedActive || started != nil {
t.Fatalf("resume finalize active=%d started=%v, want increase and NULL", active, started)
}
// Both terminal status paths fold an open segment and leave no live clock.
if _, started, err = d.StartTaskRunSegment(task.ID); err != nil {
t.Fatal(err)
} else if started == nil {
t.Fatal("terminal status setup did not open a segment")
}
if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
beforeTerminal := active
active, started, err = d.SetStatusRuntime(task.ID, "done")
if err != nil {
t.Fatal(err)
}
if active <= beforeTerminal || started != nil {
t.Fatalf("SetStatusRuntime terminal active=%d started=%v, want increase and NULL", active, started)
}
if _, _, err = d.SetStatusRuntime(task.ID, "running"); err != nil {
t.Fatal(err)
}
if _, started, err = d.StartTaskRunSegment(task.ID); err != nil {
t.Fatal(err)
} else if started == nil {
t.Fatal("guarded terminal setup did not open a segment")
}
if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
beforeGuarded := active
active, started, won, err := d.SetTerminalStatusGuardedRuntime(task.ID, "failed")
if err != nil || !won {
t.Fatalf("SetTerminalStatusGuardedRuntime: active=%d started=%v won=%v err=%v", active, started, won, err)
}
if active <= beforeGuarded || started != nil {
t.Fatalf("guarded terminal active=%d started=%v, want increase and NULL", active, started)
}
finalActive := active
active, started, won, err = d.SetTerminalStatusGuardedRuntime(task.ID, "timeout")
if err != nil || won {
t.Fatalf("second guarded terminal: active=%d started=%v won=%v err=%v", active, started, won, err)
}
if active != finalActive || started != nil {
t.Fatalf("second guarded terminal returned active=%d started=%v, want unchanged active=%d and NULL segment", active, started, finalActive)
}
got, err := d.GetTask(task.ID)
if err != nil {
t.Fatal(err)
}
if got == nil || got.ActiveRunSeconds != finalActive || got.RunSegmentStartedAt != nil {
t.Fatalf("persisted terminal runtime=%+v, want active=%d and NULL segment", got, finalActive)
}
}
func TestClearOpenTaskRunSegmentsPreservesAccumulatedRuntime(t *testing.T) {
d, err := Open(testDSN(t))
if err != nil {
t.Skipf("postgres unavailable (%v) — skipping", err)
}
defer d.Close()
task, err := d.CreateTask("runtime restart clear", "runtime restart clear test", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
if _, err := d.Exec(`UPDATE tasks SET active_run_seconds=17, run_segment_started_at=now()-interval '1 hour' WHERE id=$1`, task.ID); err != nil {
t.Fatal(err)
}
cleared, err := d.ClearOpenTaskRunSegments()
if err != nil {
t.Fatal(err)
}
if cleared < 1 {
t.Fatalf("ClearOpenTaskRunSegments cleared %d rows, want at least 1", cleared)
}
var active int64
var started sql.NullTime
if err := d.QueryRow(`SELECT active_run_seconds, run_segment_started_at FROM tasks WHERE id=$1`, task.ID).Scan(&active, &started); err != nil {
t.Fatal(err)
}
if active != 17 || started.Valid {
t.Fatalf("restart clear changed runtime: active=%d started=%v, want 17 and NULL", active, started)
}
}
+120 -17
View File
@@ -42,9 +42,11 @@ type Task struct {
CreatedAt time.Time `json:"created_at"` CreatedAt time.Time `json:"created_at"`
CompletedAt *time.Time `json:"completed_at,omitempty"` // 进入终态(done/failed/timeout)的时刻;非终态为 nil CompletedAt *time.Time `json:"completed_at,omitempty"` // 进入终态(done/failed/timeout)的时刻;非终态为 nil
// 任务级超时(见 docs/任务级超时与收尾设计.md)。 // 任务级超时(见 docs/任务级超时与收尾设计.md)。
TimeoutSeconds int `json:"timeout_seconds"` // 0=不限时 TimeoutSeconds int `json:"timeout_seconds"` // 0=不限时
FirstRunAt *time.Time `json:"first_run_at,omitempty"` // 首次真正开始运行的时刻(非 created_at);nil=尚未运行 FirstRunAt *time.Time `json:"first_run_at,omitempty"` // 首次真正开始运行的时刻(非 created_at);nil=尚未运行
DeadlineAt *time.Time `json:"deadline_at,omitempty"` // = first_run_at + timeout_seconds;nil=不限或未运行 DeadlineAt *time.Time `json:"deadline_at,omitempty"` // = first_run_at + timeout_seconds;nil=不限或未运行
ActiveRunSeconds int64 `json:"active_run_seconds"`
RunSegmentStartedAt *time.Time `json:"run_segment_started_at,omitempty"`
// planner 心跳触发间隔(秒):距上轮 plan 结束/任务开始满该值且期间无触发 → 触发一轮。 // planner 心跳触发间隔(秒):距上轮 plan 结束/任务开始满该值且期间无触发 → 触发一轮。
// 下限=默认=300(5min),低于一律抬到 300(在 CreateTask 归一)。见 docs/planner-trigger-impl-plan.md // 下限=默认=300(5min),低于一律抬到 300(在 CreateTask 归一)。见 docs/planner-trigger-impl-plan.md
PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"` PlanHeartbeatSeconds int `json:"plan_heartbeat_seconds"`
@@ -403,11 +405,11 @@ func insertTaskLLMProfiles(tx *sql.Tx, taskID int64, profileIDs []int64) error {
const taskCols = `id, COALESCE(name,''), remark, category_id, const taskCols = `id, COALESCE(name,''), remark, category_id,
COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''), COALESCE((SELECT category.name FROM task_categories category WHERE category.id=tasks.category_id),''),
description, goal, exploration_id, status, paused, queued, queued_at, COALESCE(queue_mode,''), llm_profile_id, active_llm_profile_id, COALESCE(parent_ref,''), pinned_at, created_at, completed_at, COALESCE(timeout_seconds,0), COALESCE(plan_heartbeat_seconds,300), COALESCE(coverage_enabled,true), COALESCE(worker_count,0), 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(coverage_enabled,true), COALESCE(worker_count,0), first_run_at, deadline_at, COALESCE(active_run_seconds,0), run_segment_started_at`
func scanTask(sc interface{ Scan(...any) error }) (*Task, error) { func scanTask(sc interface{ Scan(...any) error }) (*Task, error) {
var t Task var t Task
if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &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.WorkerCount, &t.FirstRunAt, &t.DeadlineAt); err != nil { if err := sc.Scan(&t.ID, &t.Name, &t.Remark, &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.WorkerCount, &t.FirstRunAt, &t.DeadlineAt, &t.ActiveRunSeconds, &t.RunSegmentStartedAt); err != nil {
return nil, err return nil, err
} }
t.Pinned = t.PinnedAt != nil t.Pinned = t.PinnedAt != nil
@@ -493,10 +495,67 @@ func (d *DB) GetTask(id int64) (*Task, error) {
// SetPaused persists a task's paused flag. // SetPaused persists a task's paused flag.
func (d *DB) SetPaused(id int64, paused bool) error { func (d *DB) SetPaused(id int64, paused bool) error {
_, err := d.Exec(`UPDATE tasks SET paused=$1 WHERE id=$2`, paused, id) _, _, err := d.SetPausedRuntime(id, paused)
return err return err
} }
// SetPausedRuntime persists a task's paused flag and, when pausing, folds any
// open execution segment into active_run_seconds. Keeping this in the DB helper
// makes direct callers safe even when they do not go through Manager pause
// orchestration.
func (d *DB) SetPausedRuntime(id int64, paused bool) (int64, *time.Time, error) {
var active int64
var started *time.Time
err := d.QueryRow(`UPDATE tasks
SET paused=$1,
active_run_seconds=active_run_seconds + CASE WHEN $1 AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=CASE WHEN $1 THEN NULL ELSE run_segment_started_at END
WHERE id=$2
RETURNING active_run_seconds, run_segment_started_at`, paused, id).Scan(&active, &started)
if err == sql.ErrNoRows {
return 0, nil, nil
}
return active, started, err
}
// StartTaskRunSegment begins the currently active execution segment. It is
// idempotent and refuses to start while a task is paused or waiting in the
// concurrency queue.
func (d *DB) StartTaskRunSegment(id int64) (int64, *time.Time, error) {
var active int64
var started *time.Time
err := d.QueryRow(`UPDATE tasks
SET run_segment_started_at=COALESCE(run_segment_started_at, now())
WHERE id=$1 AND deleted_at IS NULL AND status IN ('created','running') AND NOT paused AND NOT queued
RETURNING active_run_seconds, run_segment_started_at`, id).Scan(&active, &started)
if err == sql.ErrNoRows {
err = d.QueryRow(`SELECT active_run_seconds, run_segment_started_at FROM tasks WHERE id=$1`, id).Scan(&active, &started)
if err == sql.ErrNoRows {
return 0, nil, nil
}
}
return active, started, err
}
// FinalizeTaskRunSegment folds the current segment into active_run_seconds.
// A NULL segment makes this a safe no-op for repeated pause/terminal calls.
func (d *DB) FinalizeTaskRunSegment(id int64) (int64, *time.Time, error) {
var active int64
var started *time.Time
err := d.QueryRow(`UPDATE tasks
SET active_run_seconds=active_run_seconds + GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint),
run_segment_started_at=NULL
WHERE id=$1 AND run_segment_started_at IS NOT NULL
RETURNING active_run_seconds, run_segment_started_at`, id).Scan(&active, &started)
if err == sql.ErrNoRows {
err = d.QueryRow(`SELECT active_run_seconds, run_segment_started_at FROM tasks WHERE id=$1`, id).Scan(&active, &started)
if err == sql.ErrNoRows {
return 0, nil, nil
}
}
return active, started, err
}
// Enqueue places a task at the tail of the persistent admission queue. Repeating // Enqueue places a task at the tail of the persistent admission queue. Repeating
// the operation while it is already queued keeps its original FIFO position. // the operation while it is already queued keeps its original FIFO position.
func (d *DB) Enqueue(id int64, mode string) error { func (d *DB) Enqueue(id int64, mode string) error {
@@ -506,6 +565,8 @@ func (d *DB) Enqueue(id int64, mode string) error {
_, err := d.Exec(`UPDATE tasks _, err := d.Exec(`UPDATE tasks
SET queued=true, SET queued=true,
queued_at=CASE WHEN queued THEN COALESCE(queued_at, now()) ELSE now() END, queued_at=CASE WHEN queued THEN COALESCE(queued_at, now()) ELSE now() END,
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=NULL,
queue_mode=CASE queue_mode=CASE
WHEN queue_mode='bootstrap' OR $2='bootstrap' THEN 'bootstrap' WHEN queue_mode='bootstrap' OR $2='bootstrap' THEN 'bootstrap'
ELSE 'resume' ELSE 'resume'
@@ -539,12 +600,27 @@ func (d *DB) SetQueued(id int64, queued bool) error {
// stable); moving back to a non-terminal state clears it, so a re-run has no stale // stable); moving back to a non-terminal state clears it, so a re-run has no stale
// finish time. // finish time.
func (d *DB) SetStatus(id int64, status string) error { func (d *DB) SetStatus(id int64, status string) error {
_, err := d.Exec(` _, _, err := d.SetStatusRuntime(id, status)
return err
}
// SetStatusRuntime updates a task status and returns the committed runtime
// fields from the same UPDATE. Terminal transitions finalize any open segment.
func (d *DB) SetStatusRuntime(id int64, status string) (int64, *time.Time, error) {
var active int64
var started *time.Time
err := d.QueryRow(`
UPDATE tasks UPDATE tasks
SET status = $1, SET status = $1,
completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END,
WHERE id = $2`, status, id) active_run_seconds = active_run_seconds + CASE WHEN $1 IN ('done','failed','timeout') AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
return err run_segment_started_at = CASE WHEN $1 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
WHERE id = $2
RETURNING active_run_seconds, run_segment_started_at`, status, id).Scan(&active, &started)
if err == sql.ErrNoRows {
return 0, nil, nil
}
return active, started, err
} }
// SetTerminalStatusGuarded sets a terminal status only when the task is NOT already // SetTerminalStatusGuarded sets a terminal status only when the task is NOT already
@@ -552,16 +628,43 @@ UPDATE tasks
// Returns won=false (no error) when another terminal status already stuck — the // Returns won=false (no error) when another terminal status already stuck — the
// caller then leaves it alone. Non-terminal transitions / re-run still use SetStatus. // caller then leaves it alone. Non-terminal transitions / re-run still use SetStatus.
func (d *DB) SetTerminalStatusGuarded(id int64, status string) (won bool, err error) { func (d *DB) SetTerminalStatusGuarded(id int64, status string) (won bool, err error) {
res, err := d.Exec(` _, _, won, err = d.SetTerminalStatusGuardedRuntime(id, status)
return won, err
}
// SetTerminalStatusGuardedRuntime is the runtime-returning form used by the
// Manager so it can mirror the terminal clock before any follow-up hydration.
func (d *DB) SetTerminalStatusGuardedRuntime(id int64, status string) (active int64, started *time.Time, won bool, err error) {
err = d.QueryRow(`
UPDATE tasks UPDATE tasks
SET status = $1, SET status = $1,
completed_at = COALESCE(completed_at, now()) completed_at = COALESCE(completed_at, now()),
WHERE id = $2 AND status NOT IN ('done','failed','timeout')`, status, id) active_run_seconds = active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
if err != nil { run_segment_started_at = NULL
return false, err WHERE id = $2 AND status NOT IN ('done','failed','timeout')
RETURNING active_run_seconds, run_segment_started_at`, status, id).Scan(&active, &started)
if err == sql.ErrNoRows {
// Preserve the authoritative runtime fields for the losing terminal
// writer. The guarded transition itself is still a no-op.
err = d.QueryRow(`SELECT active_run_seconds, run_segment_started_at FROM tasks WHERE id=$1`, id).Scan(&active, &started)
if err == sql.ErrNoRows {
return 0, nil, false, nil
}
return active, started, false, err
} }
n, _ := res.RowsAffected() return active, started, err == nil, err
return n > 0, nil }
// ClearOpenTaskRunSegments drops execution segments that were live when the
// process stopped. Their wall time is unknowable after a crash or deploy, so
// startup deliberately begins with no live segment; the engine starts a fresh
// segment only when it actually resumes work.
func (d *DB) ClearOpenTaskRunSegments() (int64, error) {
res, err := d.Exec(`UPDATE tasks SET run_segment_started_at=NULL WHERE run_segment_started_at IS NOT NULL`)
if err != nil {
return 0, err
}
return res.RowsAffected()
} }
// StampFirstRun records a task's first-real-run moment and computes its absolute // StampFirstRun records a task's first-real-run moment and computes its absolute
+8 -1
View File
@@ -44,10 +44,13 @@ type TaskDTO struct {
Goal string `json:"goal"` Goal string `json:"goal"`
Status string `json:"status"` // created | running | paused | done | failed Status string `json:"status"` // created | running | paused | done | failed
CreatedAt string `json:"created_at"` CreatedAt string `json:"created_at"`
CreatedUnix int64 `json:"created_unix"` // created_at as unix seconds (for run-duration calc) CreatedUnix int64 `json:"created_unix"`
CompletedAt string `json:"completed_at"` // RFC3339 finish time (done/failed); "" if unfinished CompletedAt string `json:"completed_at"` // RFC3339 finish time (done/failed); "" if unfinished
CompletedUnix int64 `json:"completed_unix"` // completed_at as unix seconds (0 if unfinished) CompletedUnix int64 `json:"completed_unix"` // completed_at as unix seconds (0 if unfinished)
LastActivity int64 `json:"last_activity_unix"` // unix seconds of the last activity (0 if none) LastActivity int64 `json:"last_activity_unix"` // unix seconds of the last activity (0 if none)
RunSeconds int64 `json:"run_seconds"`
ActiveRunSeconds int64 `json:"active_run_seconds"`
RunSegmentStarted int64 `json:"run_segment_started_unix"`
Paused bool `json:"paused"` Paused bool `json:"paused"`
Queued bool `json:"queued"` Queued bool `json:"queued"`
Tokens TokenTotalDTO `json:"tokens"` // whole-task token consumption Tokens TokenTotalDTO `json:"tokens"` // whole-task token consumption
@@ -107,6 +110,7 @@ func tokenTotalDTO(u db.TokenUsage) TokenTotalDTO {
func taskDTO(t *Task, status string) TaskDTO { func taskDTO(t *Task, status string) TaskDTO {
lifecycle := t.lifecycleSnapshot() lifecycle := t.lifecycleSnapshot()
runSeconds := taskRunSeconds(lifecycle.ActiveRunSeconds, lifecycle.RunSegmentStartedAt, time.Now().Unix(), status, lifecycle.Paused, lifecycle.Queued)
llmState := t.llmStateSnapshot() llmState := t.llmStateSnapshot()
sourceIDs := make([]string, 0, len(lifecycle.SourceTaskIDs)) sourceIDs := make([]string, 0, len(lifecycle.SourceTaskIDs))
for _, id := range lifecycle.SourceTaskIDs { for _, id := range lifecycle.SourceTaskIDs {
@@ -127,6 +131,9 @@ func taskDTO(t *Task, status string) TaskDTO {
Status: status, Status: status,
CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)), CreatedAt: rfc3339(time.Unix(t.CreatedAt, 0)),
CreatedUnix: t.CreatedAt, CreatedUnix: t.CreatedAt,
RunSeconds: runSeconds,
ActiveRunSeconds: lifecycle.ActiveRunSeconds,
RunSegmentStarted: lifecycle.RunSegmentStartedAt,
CompletedAt: completedRFC(lifecycle.CompletedAt), CompletedAt: completedRFC(lifecycle.CompletedAt),
CompletedUnix: lifecycle.CompletedAt, CompletedUnix: lifecycle.CompletedAt,
Paused: lifecycle.Paused, Paused: lifecycle.Paused,
+6
View File
@@ -937,6 +937,9 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
return return
} }
defer e.decInflight(t.ID) defer e.decInflight(t.ID)
if err := e.m.StartTaskRunSegment(t.ID); err != nil {
log.Printf("[runtime] task %s start segment failed: %v", t.ID, err)
}
e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务) e.stampFirstRun(t) // 首次真正规划 → 盖 first_run_at + 算 deadline(仅带 timeout 的任务)
e.touch(t.ID) e.touch(t.ID)
emit := func(r db.Activity) { e.emitActivity(t, r) } emit := func(r db.Activity) { e.emitActivity(t, r) }
@@ -1086,6 +1089,9 @@ func (e *Engine) runIntent(ctx context.Context, t *Task, name string, worker *ag
} }
return true return true
} }
if err := e.m.StartTaskRunSegment(t.ID); err != nil {
log.Printf("[runtime] task %s start segment failed: %v", t.ID, err)
}
// per-work child context so the planner's kill_work can stop just this work. // per-work child context so the planner's kill_work can stop just this work.
workCtx, workCancel := context.WithCancelCause(ectx) workCtx, workCancel := context.WithCancelCause(ectx)
e.registerWork(intent.ID, workCancel) e.registerWork(intent.ID, workCancel)
+5
View File
@@ -252,6 +252,11 @@ func (s *Server) hasReadyQueuedTask(excludeID string) bool {
} }
func (s *Server) startAdmittedTask(t *Task, mode string) { func (s *Server) startAdmittedTask(t *Task, mode string) {
if s.engine.ReadyFor(t) {
if err := s.m.StartTaskRunSegment(t.ID); err != nil {
log.Printf("[runtime] task %s start segment failed: %v", t.ID, err)
}
}
if mode == "bootstrap" { if mode == "bootstrap" {
if !s.engine.beginTaskOperation(t.ID) { if !s.engine.beginTaskOperation(t.ID) {
return return
+174 -43
View File
@@ -63,6 +63,8 @@ type Task struct {
WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认 WorkerCount int `json:"worker_count"` // 本任务 Worker 数;0=用全局默认
FirstRunAt int64 `json:"first_run_at,omitempty"` FirstRunAt int64 `json:"first_run_at,omitempty"`
DeadlineAt int64 `json:"deadline_at,omitempty"` DeadlineAt int64 `json:"deadline_at,omitempty"`
ActiveRunSeconds int64 `json:"active_run_seconds"`
RunSegmentStartedAt int64 `json:"run_segment_started_at,omitempty"`
Store *pgdb.ExplorationStore `json:"-"` Store *pgdb.ExplorationStore `json:"-"`
Guard *guard.Guard `json:"-"` Guard *guard.Guard `json:"-"`
notify chan struct{} notify chan struct{}
@@ -81,21 +83,23 @@ type Task struct {
// updateLifecycle instead of reading or writing the corresponding Task fields // updateLifecycle instead of reading or writing the corresponding Task fields
// directly after the task has been published by Manager. // directly after the task has been published by Manager.
type taskLifecycleState struct { type taskLifecycleState struct {
Name string Name string
Remark string Remark string
PinnedAt int64 PinnedAt int64
Status string Status string
Paused bool Paused bool
Queued bool Queued bool
QueuedAt int64 QueuedAt int64
QueueMode string QueueMode string
CompletedAt int64 CompletedAt int64
FirstRunAt int64 FirstRunAt int64
DeadlineAt int64 DeadlineAt int64
SourceTaskIDs []int64 ActiveRunSeconds int64
CompanyIDs []int64 RunSegmentStartedAt int64
CategoryID *int64 SourceTaskIDs []int64
CategoryName string CompanyIDs []int64
CategoryID *int64
CategoryName string
} }
func (t *Task) lifecycleSnapshot() taskLifecycleState { func (t *Task) lifecycleSnapshot() taskLifecycleState {
@@ -109,21 +113,23 @@ func (t *Task) lifecycleSnapshot() taskLifecycleState {
func (t *Task) lifecycleSnapshotLocked() taskLifecycleState { func (t *Task) lifecycleSnapshotLocked() taskLifecycleState {
return taskLifecycleState{ return taskLifecycleState{
Name: t.Name, Name: t.Name,
Remark: t.Remark, Remark: t.Remark,
PinnedAt: t.PinnedAt, PinnedAt: t.PinnedAt,
Status: t.Status, Status: t.Status,
Paused: t.Paused, Paused: t.Paused,
Queued: t.Queued, Queued: t.Queued,
QueuedAt: t.QueuedAt, QueuedAt: t.QueuedAt,
QueueMode: t.QueueMode, QueueMode: t.QueueMode,
CompletedAt: t.CompletedAt, CompletedAt: t.CompletedAt,
FirstRunAt: t.FirstRunAt, FirstRunAt: t.FirstRunAt,
DeadlineAt: t.DeadlineAt, DeadlineAt: t.DeadlineAt,
SourceTaskIDs: append([]int64(nil), t.SourceTaskIDs...), ActiveRunSeconds: t.ActiveRunSeconds,
CompanyIDs: append([]int64(nil), t.CompanyIDs...), RunSegmentStartedAt: t.RunSegmentStartedAt,
CategoryID: cloneInt64Ptr(t.CategoryID), SourceTaskIDs: append([]int64(nil), t.SourceTaskIDs...),
CategoryName: t.CategoryName, CompanyIDs: append([]int64(nil), t.CompanyIDs...),
CategoryID: cloneInt64Ptr(t.CategoryID),
CategoryName: t.CategoryName,
} }
} }
@@ -145,6 +151,8 @@ func (t *Task) updateLifecycle(update func(*taskLifecycleState)) {
t.CompletedAt = state.CompletedAt t.CompletedAt = state.CompletedAt
t.FirstRunAt = state.FirstRunAt t.FirstRunAt = state.FirstRunAt
t.DeadlineAt = state.DeadlineAt t.DeadlineAt = state.DeadlineAt
t.ActiveRunSeconds = state.ActiveRunSeconds
t.RunSegmentStartedAt = state.RunSegmentStartedAt
t.SourceTaskIDs = append(t.SourceTaskIDs[:0], state.SourceTaskIDs...) t.SourceTaskIDs = append(t.SourceTaskIDs[:0], state.SourceTaskIDs...)
t.CompanyIDs = append(t.CompanyIDs[:0], state.CompanyIDs...) t.CompanyIDs = append(t.CompanyIDs[:0], state.CompanyIDs...)
t.CategoryID = cloneInt64Ptr(state.CategoryID) t.CategoryID = cloneInt64Ptr(state.CategoryID)
@@ -525,6 +533,13 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
// A live segment cannot be trusted across a process restart: its end time
// could be deployment/crash downtime. Clear it before tasks are restored so
// the next engine admission starts a fresh segment at actual execution time.
if _, err := pg.ClearOpenTaskRunSegments(); err != nil {
_ = pg.Close()
return nil, fmt.Errorf("clear stale task run segments: %w", err)
}
if err := pg.RecoverFindingRetests(); err != nil { if err := pg.RecoverFindingRetests(); err != nil {
pg.Close() pg.Close()
return nil, fmt.Errorf("recover finding retests: %w", err) return nil, fmt.Errorf("recover finding retests: %w", err)
@@ -1009,6 +1024,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds, TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds,
CoverageEnabled: pt.CoverageEnabled, WorkerCount: pt.WorkerCount, CoverageEnabled: pt.CoverageEnabled, WorkerCount: pt.WorkerCount,
FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt), FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt),
ActiveRunSeconds: pt.ActiveRunSeconds, RunSegmentStartedAt: unixOrZero(pt.RunSegmentStartedAt),
Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1), Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1),
} }
} }
@@ -1289,13 +1305,16 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error {
if err != nil { if err != nil {
return err return err
} }
if err := m.pg.SetPaused(n, paused); err != nil { active, started, err := m.pg.SetPausedRuntime(n, paused)
if err != nil {
return err return err
} }
m.mu.Lock() m.mu.Lock()
if t := m.tasks[id]; t != nil { if t := m.tasks[id]; t != nil {
t.updateLifecycle(func(state *taskLifecycleState) { t.updateLifecycle(func(state *taskLifecycleState) {
state.Paused = paused state.Paused = paused
state.ActiveRunSeconds = active
state.RunSegmentStartedAt = unixOrZero(started)
}) })
} }
m.mu.Unlock() m.mu.Unlock()
@@ -1331,7 +1350,8 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
mode = "" mode = ""
} }
var queuedAt, completedAt, firstRunAt, deadlineAt sql.NullTime var queuedAt, completedAt, firstRunAt, deadlineAt, segmentStartedAt sql.NullTime
var activeRunSeconds int64
var committedMode string var committedMode string
err = m.pg.QueryRow(`UPDATE tasks err = m.pg.QueryRow(`UPDATE tasks
SET status=$2, SET status=$2,
@@ -1360,9 +1380,11 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
WHEN $6='timeout' AND $2 NOT IN ('done','failed','timeout') THEN NULL WHEN $6='timeout' AND $2 NOT IN ('done','failed','timeout') THEN NULL
ELSE deadline_at ELSE deadline_at
END END
,active_run_seconds=active_run_seconds + CASE WHEN ($3 OR $2 IN ('done','failed','timeout')) AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END
,run_segment_started_at=CASE WHEN $3 OR $2 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END
WHERE id=$1 AND deleted_at IS NULL AND status=$6 WHERE id=$1 AND deleted_at IS NULL AND status=$6
RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil). RETURNING queued_at, queue_mode, completed_at, first_run_at, deadline_at, active_run_seconds, run_segment_started_at`, n, status, queued, mode, preservePosition, expectedStatus, manualUntil).
Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt) Scan(&queuedAt, &committedMode, &completedAt, &firstRunAt, &deadlineAt, &activeRunSeconds, &segmentStartedAt)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus) return fmt.Errorf("task %s lifecycle changed before admission (expected status %q)", id, expectedStatus)
} }
@@ -1393,6 +1415,11 @@ func (m *Manager) applyTaskAdmission(id, expectedStatus, status string, queued b
if deadlineAt.Valid { if deadlineAt.Valid {
state.DeadlineAt = deadlineAt.Time.Unix() state.DeadlineAt = deadlineAt.Time.Unix()
} }
state.ActiveRunSeconds = activeRunSeconds
state.RunSegmentStartedAt = 0
if segmentStartedAt.Valid {
state.RunSegmentStartedAt = segmentStartedAt.Time.Unix()
}
}) })
} }
m.mu.Unlock() m.mu.Unlock()
@@ -1413,11 +1440,15 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
return err return err
} }
var mode string var mode string
var activeRunSeconds int64
var segmentStartedAt sql.NullTime
err = m.pg.QueryRow(`UPDATE tasks err = m.pg.QueryRow(`UPDATE tasks
SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL SET paused=true, queued=false, queued_at=NULL, schedule_paused=$2, schedule_manual_until=NULL,
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=NULL
WHERE id=$1 AND deleted_at IS NULL AND paused=false WHERE id=$1 AND deleted_at IS NULL AND paused=false
AND status NOT IN ('done','failed','timeout') AND status NOT IN ('done','failed','timeout')
RETURNING COALESCE(queue_mode,'')`, n, scheduled).Scan(&mode) RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n, scheduled).Scan(&mode, &activeRunSeconds, &segmentStartedAt)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("task %s is unavailable for pause", id) return fmt.Errorf("task %s is unavailable for pause", id)
} }
@@ -1431,6 +1462,11 @@ func (m *Manager) ApplyTaskPauseOrigin(id string, scheduled bool) error {
state.Queued = false state.Queued = false
state.QueuedAt = 0 state.QueuedAt = 0
state.QueueMode = mode state.QueueMode = mode
state.ActiveRunSeconds = activeRunSeconds
state.RunSegmentStartedAt = 0
if segmentStartedAt.Valid {
state.RunSegmentStartedAt = segmentStartedAt.Time.Unix()
}
}) })
} }
m.mu.Unlock() m.mu.Unlock()
@@ -1449,16 +1485,20 @@ func (m *Manager) EnqueueTask(id, mode string) error {
return fmt.Errorf("invalid queue mode %q", mode) return fmt.Errorf("invalid queue mode %q", mode)
} }
var queuedAt time.Time var queuedAt time.Time
var activeRunSeconds int64
var segmentStartedAt sql.NullTime
var committedMode string var committedMode string
err = m.pg.QueryRow(`UPDATE tasks err = m.pg.QueryRow(`UPDATE tasks
SET queued=true, SET queued=true,
queued_at=CASE WHEN queued THEN COALESCE(queued_at, now()) ELSE now() END, queued_at=CASE WHEN queued THEN COALESCE(queued_at, now()) ELSE now() END,
active_run_seconds=active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END,
run_segment_started_at=NULL,
queue_mode=CASE queue_mode=CASE
WHEN queue_mode='bootstrap' OR $2='bootstrap' THEN 'bootstrap' WHEN queue_mode='bootstrap' OR $2='bootstrap' THEN 'bootstrap'
ELSE 'resume' ELSE 'resume'
END END
WHERE id=$1 AND deleted_at IS NULL WHERE id=$1 AND deleted_at IS NULL
RETURNING queued_at, queue_mode`, n, mode).Scan(&queuedAt, &committedMode) RETURNING queued_at, queue_mode, active_run_seconds, run_segment_started_at`, n, mode).Scan(&queuedAt, &committedMode, &activeRunSeconds, &segmentStartedAt)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("task %s is unavailable for enqueue", id) return fmt.Errorf("task %s is unavailable for enqueue", id)
} }
@@ -1471,6 +1511,80 @@ func (m *Manager) EnqueueTask(id, mode string) error {
state.QueuedAt = queuedAt.UnixNano() state.QueuedAt = queuedAt.UnixNano()
state.Queued = true state.Queued = true
state.QueueMode = committedMode state.QueueMode = committedMode
state.ActiveRunSeconds = activeRunSeconds
state.RunSegmentStartedAt = 0
if segmentStartedAt.Valid {
state.RunSegmentStartedAt = segmentStartedAt.Time.Unix()
}
})
}
m.mu.Unlock()
return nil
}
// StartTaskRunSegment records the start of actual execution, once the engine
// has admitted a task and is about to run planner/worker work.
func (m *Manager) StartTaskRunSegment(id string) error {
m.taskStateMu.Lock()
defer m.taskStateMu.Unlock()
n, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return err
}
m.mu.RLock()
if t := m.tasks[id]; t != nil && t.lifecycleSnapshot().RunSegmentStartedAt > 0 {
m.mu.RUnlock()
return nil
}
m.mu.RUnlock()
active, started, err := m.pg.StartTaskRunSegment(n)
if err != nil {
return err
}
if started != nil {
m.mirrorTaskRuntime(id, active, started)
}
return nil
}
// FinalizeTaskRunSegment is idempotent and is used by pause/terminal paths that
// may race with a late engine cancellation.
func (m *Manager) FinalizeTaskRunSegment(id string) error {
m.taskStateMu.Lock()
defer m.taskStateMu.Unlock()
n, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return err
}
active, started, err := m.pg.FinalizeTaskRunSegment(n)
if err != nil {
return err
}
m.mirrorTaskRuntime(id, active, started)
return nil
}
func (m *Manager) mirrorTaskRuntime(id string, active int64, started *time.Time) {
m.mu.Lock()
defer m.mu.Unlock()
if t := m.tasks[id]; t != nil {
t.updateLifecycle(func(state *taskLifecycleState) {
state.ActiveRunSeconds = active
state.RunSegmentStartedAt = unixOrZero(started)
})
}
}
func (m *Manager) syncTaskRuntime(id string, n int64) error {
pt, err := m.pg.GetTask(n)
if err != nil || pt == nil {
return err
}
m.mu.Lock()
if t := m.tasks[id]; t != nil {
t.updateLifecycle(func(state *taskLifecycleState) {
state.ActiveRunSeconds = pt.ActiveRunSeconds
state.RunSegmentStartedAt = unixOrZero(pt.RunSegmentStartedAt)
}) })
} }
m.mu.Unlock() m.mu.Unlock()
@@ -1559,20 +1673,30 @@ func (m *Manager) SetTaskStatusGuarded(id, status string) (won bool, err error)
if err != nil { if err != nil {
return false, err return false, err
} }
won, err = m.pg.SetTerminalStatusGuarded(n, status) active, started, won, err := m.pg.SetTerminalStatusGuardedRuntime(n, status)
if err != nil || !won { if err != nil || !won {
return won, err return won, err
} }
// Mirror the committed terminal transition before any optional hydration.
// This keeps the in-memory clock stopped even if GetTask later fails.
m.mu.Lock() m.mu.Lock()
if t := m.tasks[id]; t != nil { if t := m.tasks[id]; t != nil {
t.updateLifecycle(func(state *taskLifecycleState) { t.updateLifecycle(func(state *taskLifecycleState) {
state.Status = status state.Status = status
if state.CompletedAt == 0 { state.ActiveRunSeconds = active
state.CompletedAt = time.Now().Unix() state.RunSegmentStartedAt = unixOrZero(started)
if pgdb.IsTerminal(status) {
state.RunSegmentStartedAt = 0
if state.CompletedAt == 0 {
state.CompletedAt = time.Now().Unix()
}
} }
}) })
} }
m.mu.Unlock() m.mu.Unlock()
if err := m.syncTaskRuntime(id, n); err != nil {
return true, err
}
return true, nil return true, nil
} }
@@ -1585,16 +1709,20 @@ func (m *Manager) SetTaskStatus(id, status string) error {
if err != nil { if err != nil {
return err return err
} }
if err := m.pg.SetStatus(n, status); err != nil { active, started, err := m.pg.SetStatusRuntime(n, status)
if err != nil {
return err return err
} }
// Mirror the successful UPDATE immediately. A later GetTask hydration is
// best-effort and must not reopen a terminal clock in memory.
m.mu.Lock() m.mu.Lock()
if t := m.tasks[id]; t != nil { if t := m.tasks[id]; t != nil {
t.updateLifecycle(func(state *taskLifecycleState) { t.updateLifecycle(func(state *taskLifecycleState) {
state.Status = status state.Status = status
// Mirror the DB's completed_at stamp on the live handle so the DTO shows state.ActiveRunSeconds = active
// the finish time without a reload (terminal -> stamp once; else clear). state.RunSegmentStartedAt = unixOrZero(started)
if pgdb.IsTerminal(status) { if pgdb.IsTerminal(status) {
state.RunSegmentStartedAt = 0
if state.CompletedAt == 0 { if state.CompletedAt == 0 {
state.CompletedAt = time.Now().Unix() state.CompletedAt = time.Now().Unix()
} }
@@ -1604,6 +1732,9 @@ func (m *Manager) SetTaskStatus(id, status string) error {
}) })
} }
m.mu.Unlock() m.mu.Unlock()
if err := m.syncTaskRuntime(id, n); err != nil {
return err
}
return nil return nil
} }
+3 -12
View File
@@ -175,10 +175,9 @@ func (s *Server) delegateToTask(ctx context.Context, in json.RawMessage, pick fu
func (s *Server) toolListTasks() actool.CoreTool { func (s *Server) toolListTasks() actool.CoreTool {
return roTool("list_tasks", return roTool("list_tasks",
"列出所有任务(id/描述/目标/状态/运行时长/父任务/LLM 配置),编排 agent 用它掌握全局、看哪些任务卡太久、各自用哪个 LLM。运行时长:运行中=创建→现在,终态=创建→最后活动(秒)。llm_profile:任务 planner/worker 用的配置名,(激活配置)=跟随全局激活。", "列出所有任务(id/描述/目标/状态/运行时长/父任务/LLM 配置),编排 agent 用它掌握全局、看哪些任务卡太久、各自用哪个 LLM。运行时长只累计实际执行时间,暂停、排队和终态后的时间不计入。llm_profile:任务 planner/worker 用的配置名,(激活配置)=跟随全局激活。",
objSchema(map[string]any{}), objSchema(map[string]any{}),
func(context.Context, json.RawMessage) (actool.Result, error) { func(context.Context, json.RawMessage) (actool.Result, error) {
lastAct, _ := s.m.PG().LastActivityAll()
// id -> name to resolve each task's pinned LLM profile. // id -> name to resolve each task's pinned LLM profile.
profName := map[int64]string{} profName := map[int64]string{}
if profs, err := s.m.pg.ListProfiles(); err == nil { if profs, err := s.m.pg.ListProfiles(); err == nil {
@@ -189,16 +188,8 @@ func (s *Server) toolListTasks() actool.CoreTool {
out := make([]map[string]any, 0) out := make([]map[string]any, 0)
for _, t := range s.m.List() { for _, t := range s.m.List() {
status := s.deriveTaskStatus(t) status := s.deriveTaskStatus(t)
end := lastAct[t.ExpID] lifecycle := t.lifecycleSnapshot()
if live := s.engine.LastActivity(t.ID); live > end { dur := taskRunSeconds(lifecycle.ActiveRunSeconds, lifecycle.RunSegmentStartedAt, time.Now().Unix(), status, lifecycle.Paused, lifecycle.Queued)
end = live
}
dur := int64(0)
if status == "running" {
dur = time.Now().Unix() - t.CreatedAt
} else if end > t.CreatedAt {
dur = end - t.CreatedAt
}
row := map[string]any{"id": t.ID, "description": t.Description, "goal": t.Goal, "status": status, "run_seconds": dur} row := map[string]any{"id": t.ID, "description": t.Description, "goal": t.Goal, "status": status, "run_seconds": dur}
if t.ParentRef != "" { if t.ParentRef != "" {
row["parent_ref"] = t.ParentRef row["parent_ref"] = t.ParentRef
+1 -1
View File
@@ -1138,7 +1138,7 @@ func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) {
metric := metrics[t.ExpID] metric := metrics[t.ExpID]
dto.Tokens = tokenTotalDTO(metric.Tokens) dto.Tokens = tokenTotalDTO(metric.Tokens)
// prefer the live in-memory heartbeat (fresher) and fall back to the // prefer the live in-memory heartbeat (fresher) and fall back to the
// persisted max activity time (survives restarts) for run-duration display. // persisted max activity time (survives restarts) for activity display.
dto.LastActivity = metric.LastActivity dto.LastActivity = metric.LastActivity
if live := s.engine.LastActivity(t.ID); live > dto.LastActivity { if live := s.engine.LastActivity(t.ID); live > dto.LastActivity {
dto.LastActivity = live dto.LastActivity = live
+25
View File
@@ -0,0 +1,25 @@
package server
// taskRunSeconds computes displayed active runtime from persisted segment data.
// Segment timestamps are unix seconds; a zero start means no active segment.
func taskRunSeconds(active, segmentStarted, now int64, status string, paused, queued bool) int64 {
if active < 0 {
active = 0
}
if status == "running" && !paused && !queued && segmentStarted > 0 && now > segmentStarted {
return active + now - segmentStarted
}
return active
}
// finalizeTaskRunSeconds mirrors the database transition for deterministic unit
// tests and callers that need to reason about a segment without a live database.
func finalizeTaskRunSeconds(active, segmentStarted, now int64) (int64, int64) {
if active < 0 {
active = 0
}
if segmentStarted > 0 && now > segmentStarted {
active += now - segmentStarted
}
return active, 0
}
+35
View File
@@ -0,0 +1,35 @@
package server
import "testing"
func TestTaskRunSecondsRunningSegmentAccumulates(t *testing.T) {
if got := taskRunSeconds(12, 100, 145, "running", false, false); got != 57 {
t.Fatalf("running duration = %d, want 57", got)
}
}
func TestTaskRunSecondsPausedDoesNotGrow(t *testing.T) {
if got := taskRunSeconds(57, 100, 1000, "paused", true, false); got != 57 {
t.Fatalf("paused duration = %d, want 57", got)
}
if got := taskRunSeconds(57, 100, 1000, "running", false, true); got != 57 {
t.Fatalf("queued duration = %d, want 57", got)
}
}
func TestTaskRunSecondsResumeContinuesFromPriorTotal(t *testing.T) {
active, segment := finalizeTaskRunSeconds(12, 100, 145)
if active != 57 || segment != 0 {
t.Fatalf("finalized segment = (%d,%d), want (57,0)", active, segment)
}
if got := taskRunSeconds(active, 200, 230, "running", false, false); got != 87 {
t.Fatalf("resumed duration = %d, want 87", got)
}
}
func TestTaskRunSecondsTerminalFinalizesSegment(t *testing.T) {
active, segment := finalizeTaskRunSeconds(20, 500, 560)
if active != 80 || segment != 0 {
t.Fatalf("terminal finalization = (%d,%d), want (80,0)", active, segment)
}
}
+8 -12
View File
@@ -188,19 +188,15 @@ function fmtDuration(sec: number): string {
return `${s}s`; return `${s}s`;
} }
// taskDuration is the task's run duration in seconds: created → now while running, // taskDuration counts only active execution segments. A live segment ticks locally;
// created → completion for a finished task, else created → last activity. 0 when it // paused, queued, and terminal tasks stay fixed at their persisted active total.
// never ran (no activity yet).
function taskDuration(task: Task, nowSec: number): number { function taskDuration(task: Task, nowSec: number): number {
const start = task.created_unix ?? 0; const active = Math.max(0, Math.floor(task.active_run_seconds ?? 0));
if (!start) return 0; const segmentStart = task.run_segment_started_unix ?? 0;
const end = if (task.status === "running" && !task.paused && !task.queued && segmentStart > 0) {
task.status === "running" return active + Math.max(0, Math.floor(nowSec - segmentStart));
? nowSec }
: task.completed_unix && task.completed_unix > 0 return active;
? task.completed_unix
: (task.last_activity_unix ?? 0);
return end > start ? end - start : 0;
} }
// deleteDetails takes only the countable part of the result so callers can pass an // deleteDetails takes only the countable part of the result so callers can pass an
+12
View File
@@ -60,6 +60,9 @@ export const tasks: Task[] = [
status: "running", status: "running",
created_at: T("2026-07-24T09:12:00Z"), created_at: T("2026-07-24T09:12:00Z"),
created_unix: 1785489120, created_unix: 1785489120,
active_run_seconds: 28800,
run_segment_started_unix: 1785643102,
run_seconds: 28800,
last_activity: T("2026-07-26T03:58:22Z"), last_activity: T("2026-07-26T03:58:22Z"),
last_activity_unix: 1785643102, last_activity_unix: 1785643102,
paused: false, paused: false,
@@ -86,6 +89,9 @@ export const tasks: Task[] = [
status: "running", status: "running",
created_at: T("2026-07-25T14:05:00Z"), created_at: T("2026-07-25T14:05:00Z"),
created_unix: 1785592500, created_unix: 1785592500,
active_run_seconds: 14400,
run_segment_started_unix: 1785642010,
run_seconds: 14400,
last_activity: T("2026-07-26T03:40:10Z"), last_activity: T("2026-07-26T03:40:10Z"),
last_activity_unix: 1785642010, last_activity_unix: 1785642010,
paused: false, paused: false,
@@ -112,6 +118,9 @@ export const tasks: Task[] = [
status: "paused", status: "paused",
created_at: T("2026-07-22T08:30:00Z"), created_at: T("2026-07-22T08:30:00Z"),
created_unix: 1785313800, created_unix: 1785313800,
active_run_seconds: 21600,
run_segment_started_unix: 0,
run_seconds: 21600,
last_activity: T("2026-07-23T18:40:10Z"), last_activity: T("2026-07-23T18:40:10Z"),
last_activity_unix: 1785436810, last_activity_unix: 1785436810,
paused: true, paused: true,
@@ -137,6 +146,9 @@ export const tasks: Task[] = [
status: "done", status: "done",
created_at: T("2026-07-18T08:00:00Z"), created_at: T("2026-07-18T08:00:00Z"),
created_unix: 1784966400, created_unix: 1784966400,
active_run_seconds: 86400,
run_segment_started_unix: 0,
run_seconds: 86400,
completed_at: T("2026-07-20T22:11:00Z"), completed_at: T("2026-07-20T22:11:00Z"),
completed_unix: 1785190260, completed_unix: 1785190260,
last_activity: T("2026-07-20T22:11:00Z"), last_activity: T("2026-07-20T22:11:00Z"),
+9
View File
@@ -951,6 +951,11 @@ function controlMockTask(id: string, action: "pause" | "resume"): BatchControlIt
task.queued = false; task.queued = false;
task.status = "paused"; task.status = "paused";
task.engine_mode = "paused"; task.engine_mode = "paused";
if (task.run_segment_started_unix && task.run_segment_started_unix > 0) {
task.active_run_seconds = (task.active_run_seconds ?? 0) + Math.max(0, Math.floor(Date.now() / 1000) - task.run_segment_started_unix);
task.run_seconds = task.active_run_seconds;
}
task.run_segment_started_unix = 0;
} else { } else {
if (!task.paused && task.status !== "paused") { if (!task.paused && task.status !== "paused") {
return { id, ok: false, status: task.status, error: "任务未暂停" }; return { id, ok: false, status: task.status, error: "任务未暂停" };
@@ -959,6 +964,7 @@ function controlMockTask(id: string, action: "pause" | "resume"): BatchControlIt
task.queued = false; task.queued = false;
task.status = "running"; task.status = "running";
task.engine_mode = "exploring"; task.engine_mode = "exploring";
task.run_segment_started_unix = Math.floor(Date.now() / 1000);
} }
return { id, ok: true, status: task.status, queued: false }; return { id, ok: true, status: task.status, queued: false };
} }
@@ -1248,6 +1254,9 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re
status: "created", status: "created",
created_at: now.toISOString(), created_at: now.toISOString(),
created_unix: Math.floor(now.getTime() / 1000), created_unix: Math.floor(now.getTime() / 1000),
active_run_seconds: 0,
run_segment_started_unix: 0,
run_seconds: 0,
paused: false, paused: false,
active: true, active: true,
in_flight: 0, in_flight: 0,
+4 -1
View File
@@ -34,10 +34,13 @@ export interface Task {
goal: string; goal: string;
status: TaskStatus; status: TaskStatus;
created_at: string; created_at: string;
created_unix?: number; // created_at as unix seconds (run-duration calc) created_unix?: number;
completed_at?: string; // RFC3339 finish time (done/failed); "" if unfinished completed_at?: string; // RFC3339 finish time (done/failed); "" if unfinished
completed_unix?: number; // completed_at as unix seconds (0/undef if unfinished) completed_unix?: number; // completed_at as unix seconds (0/undef if unfinished)
last_activity_unix?: number; // unix seconds of the last activity (0/undef if none) last_activity_unix?: number; // unix seconds of the last activity (0/undef if none)
run_seconds?: number; // backend-computed active runtime seconds
active_run_seconds?: number; // completed running segments in seconds
run_segment_started_unix?: number; // current running segment start; 0 when inactive
paused?: boolean; paused?: boolean;
queued?: boolean; queued?: boolean;
active?: boolean; active?: boolean;