From cca12ab08a45f60e5736d8709e8aef453834df18 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 26 Feb 2026 19:14:47 +0800 Subject: [PATCH] feat(audit): auto retry missed tasks Add retry-based auto remediation for audit findings and document new config knobs. --- README.md | 35 +++++ README.zh.md | 35 +++++ config/config.example.json | 3 + docs/design/orchestration-audit.md | 51 ++++++- pkg/agent/audit.go | 211 +++++++++++++++++++++++++++-- pkg/agent/audit_test.go | 99 ++++++++++++++ pkg/agent/instance.go | 1 + pkg/agent/loop.go | 1 + pkg/config/config.go | 11 +- pkg/config/defaults.go | 17 ++- 10 files changed, 446 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 67e9fd8c8..9614369c0 100644 --- a/README.md +++ b/README.md @@ -785,6 +785,41 @@ The subagent has access to tools (message, web_search, etc.) and can communicate * `PICOCLAW_HEARTBEAT_ENABLED=false` to disable * `PICOCLAW_HEARTBEAT_INTERVAL=60` to change interval +### Audit (Task Supervisor) + +PicoClaw can periodically audit background tasks (subagents spawned via `spawn/sessions_spawn`) and detect: + +- missed tasks (overdue planned/running, failed with retry budget) +- low-quality completions (empty result) +- execution inconsistencies (no tool evidence in strict mode) + +You can optionally enable **automatic make-up / retry** by spawning remediation subagents: + +- `safe_only` (default): record remediation notes only +- `retry_missed`: auto-retry missed findings +- `retry_all`: auto-retry missed + quality + inconsistency + +**Example configuration:** + +```json +{ + "orchestration": { + "retry_limit_per_task": 10 + }, + "audit": { + "enabled": true, + "interval_minutes": 5, + "lookback_minutes": 180, + "auto_remediation": "retry_all", + "max_auto_remediations_per_cycle": 3, + "remediation_cooldown_minutes": 10, + "notify_channel": "last_active" + } +} +``` + +> Tip: Leave `audit.remediation_agent_id` empty to run retries on the same model. Set it if you want retries to run on a dedicated supervisor agent. + ### Providers > [!NOTE] diff --git a/README.zh.md b/README.zh.md index 98c8ffa4b..6420ca871 100644 --- a/README.zh.md +++ b/README.zh.md @@ -420,6 +420,41 @@ Agent 读取 HEARTBEAT.md - `PICOCLAW_HEARTBEAT_ENABLED=false` 禁用 - `PICOCLAW_HEARTBEAT_INTERVAL=60` 更改间隔 +### 审计 / 监管与自动补做 (Audit) + +PicoClaw 支持周期性审计后台任务（主要是通过 `spawn/sessions_spawn` 生成的子任务），用于发现： + +- **漏执行 / 超时**：planned/running 任务超时、failed 但仍有重试预算 +- **低质量输出**：completed 但结果为空 +- **执行不一致**：strict 模式下 completed 但没有工具证据 + +你也可以开启**自动补做 / 自动重试**（通过再 spawn 一个补做 subagent）： + +- `safe_only`（默认）：仅记录提示，不自动重试 +- `retry_missed`：自动补做 missed +- `retry_all`：自动补做 missed + quality + inconsistency + +**配置示例：** + +```json +{ + "orchestration": { + "retry_limit_per_task": 10 + }, + "audit": { + "enabled": true, + "interval_minutes": 5, + "lookback_minutes": 180, + "auto_remediation": "retry_all", + "max_auto_remediations_per_cycle": 3, + "remediation_cooldown_minutes": 10, + "notify_channel": "last_active" + } +} +``` + +> 小贴士：保持 `audit.remediation_agent_id` 为空即可让补做任务使用同一个模型；如果你想把补做任务交给一个专门的监管 agent，再配置该字段即可。 + ### 提供商 (Providers) > [!NOTE] diff --git a/config/config.example.json b/config/config.example.json index 457a9b505..4b6fd3050 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -296,6 +296,9 @@ "min_confidence": 0.75, "inconsistency_policy": "strict", "auto_remediation": "safe_only", + "max_auto_remediations_per_cycle": 3, + "remediation_cooldown_minutes": 10, + "remediation_agent_id": "", "notify_channel": "last_active", "supervisor": { "enabled": false, diff --git a/docs/design/orchestration-audit.md b/docs/design/orchestration-audit.md index 2e97b2403..c22934035 100644 --- a/docs/design/orchestration-audit.md +++ b/docs/design/orchestration-audit.md @@ -40,6 +40,7 @@ This document describes how PicoClaw's subagent orchestration and periodic audit - Used as default deadline metadata in task ledger entries. - `retry_limit_per_task` - Used by audit logic to detect failed tasks that still have retry budget. + - Also caps how many automatic remediation retries can be spawned for one task. ### `audit` @@ -54,7 +55,16 @@ This document describes how PicoClaw's subagent orchestration and periodic audit - `inconsistency_policy` - `strict` mode flags completed tasks with no tool evidence. - `auto_remediation` - - `safe_only` records low-risk remediation actions in ledger. + - `safe_only` records low-risk remediation actions in ledger (no retries). + - `retry_missed` automatically spawns subagent retries for `missed` findings. + - `retry_all` automatically spawns retries for `missed`, `quality`, and `inconsistency` findings. + - `retry` is accepted as an alias for `retry_missed`. +- `max_auto_remediations_per_cycle` + - Caps how many remediation tasks can be spawned in one audit cycle (prevents runaway loops). +- `remediation_cooldown_minutes` + - Suppresses repeated remediation attempts for the same task within the cooldown window. +- `remediation_agent_id` + - Optional agent id used to execute remediation retries (requires subagent allowlist when targeting a different agent). - `notify_channel` - Destination for audit report: - `last_active`: last recorded user channel/chat. @@ -118,6 +128,45 @@ Optional model checks: - Supervisor model receives task JSON and returns structured score/issues. - Findings are merged into deterministic report. +## Auto Remediation (Retry / Make-up) + +When `audit.auto_remediation` is set to a retry mode (`retry_missed` / `retry_all`), the audit loop can automatically **spawn a subagent** to retry missed or low-quality tasks. + +Behavior: + +- The audit loop scans the task ledger for findings. +- For eligible findings, it spawns a new subagent task with stricter acceptance criteria. +- It records a remediation entry (`action=retry`) and increments `retry_count` on the original task. +- It respects: + - `audit.max_auto_remediations_per_cycle` (per-cycle cap), + - `audit.remediation_cooldown_minutes` (per-task cooldown), + - `orchestration.retry_limit_per_task` (per-task retry budget). + +Delivery: + +- By default, remediation retries run on the default agent using the same provider/model as normal subagent tasks. +- If `audit.remediation_agent_id` is set, the retry is delegated to that agent id (requires subagent allowlist when targeting a different agent). +- The retry result is delivered back to the original task's `origin_channel/origin_chat_id` when available; otherwise it falls back to `audit.notify_channel`. + +### Example Configuration + +```json +{ + "orchestration": { + "retry_limit_per_task": 10 + }, + "audit": { + "enabled": true, + "interval_minutes": 5, + "lookback_minutes": 180, + "auto_remediation": "retry_all", + "max_auto_remediations_per_cycle": 3, + "remediation_cooldown_minutes": 10, + "notify_channel": "last_active" + } +} +``` + ## Backward Compatibility - Existing tools and loop behavior remain unchanged when `audit.enabled=false`. diff --git a/pkg/agent/audit.go b/pkg/agent/audit.go index bec87083a..6177488fa 100644 --- a/pkg/agent/audit.go +++ b/pkg/agent/audit.go @@ -128,7 +128,7 @@ func (al *AgentLoop) executeAuditCycle(ctx context.Context) { return } - al.applyAutoRemediation(report) + al.applyAutoRemediation(ctx, report) al.publishAuditReport(report) } @@ -369,8 +369,8 @@ func parseSupervisorReview(raw string) (*supervisorReview, error) { return &review, nil } -func (al *AgentLoop) applyAutoRemediation(report *AuditReport) { - if report == nil || len(report.Findings) == 0 || al.taskLedger == nil { +func (al *AgentLoop) applyAutoRemediation(ctx context.Context, report *AuditReport) { + if report == nil || len(report.Findings) == 0 || al.taskLedger == nil || al.cfg == nil { return } @@ -378,22 +378,215 @@ func (al *AgentLoop) applyAutoRemediation(report *AuditReport) { if mode == "" || mode == "disabled" || mode == "off" || mode == "none" { return } - if mode != "safe_only" { + + if mode == "safe_only" { + for _, finding := range report.Findings { + if finding.Category != "missed" { + continue + } + _ = al.taskLedger.AddRemediation(finding.TaskID, tools.TaskRemediation{ + Action: "notify", + Status: "queued", + Note: finding.Message, + }) + } return } + // retry/auto-fix modes + // - retry_missed: only retry tasks in "missed" category + // - retry_all: retry missed + rerun quality/inconsistency findings + // - retry: alias for retry_missed + switch mode { + case "retry": + mode = "retry_missed" + case "retry_missed", "retry_all": + default: + // Unknown mode: fail closed. + return + } + + maxPerCycle := al.cfg.Audit.MaxAutoRemediationsPerCycle + if maxPerCycle <= 0 { + maxPerCycle = 3 + } + cooldownMinutes := al.cfg.Audit.RemediationCooldownMinutes + if cooldownMinutes <= 0 { + cooldownMinutes = 10 + } + cooldownMS := int64(cooldownMinutes) * 60 * 1000 + retryLimit := al.cfg.Orchestration.RetryLimitPerTask + if retryLimit < 0 { + retryLimit = 0 + } + + targetAgentID := strings.TrimSpace(al.cfg.Audit.RemediationAgentID) + defaultAgent := al.registry.GetDefaultAgent() + if defaultAgent == nil || defaultAgent.SubagentManager == nil { + return + } + if targetAgentID != "" { + normalized, ok := al.registry.GetAgent(targetAgentID) + if !ok || normalized == nil { + logger.WarnCF("audit", "Remediation agent not found; falling back to default agent", map[string]any{ + "remediation_agent_id": targetAgentID, + }) + targetAgentID = "" + } else { + targetAgentID = normalized.ID + } + if targetAgentID != "" && targetAgentID != defaultAgent.ID && !al.registry.CanSpawnSubagent(defaultAgent.ID, targetAgentID) { + logger.WarnCF("audit", "Remediation agent not allowed by subagent allowlist; falling back to default agent", map[string]any{ + "parent_agent_id": defaultAgent.ID, + "remediation_agent_id": targetAgentID, + }) + targetAgentID = "" + } + } + + nowMS := time.Now().UnixMilli() + thresholdMS := nowMS - cooldownMS + spawned := 0 for _, finding := range report.Findings { - if finding.Category != "missed" { + if spawned >= maxPerCycle { + return + } + + if strings.TrimSpace(finding.TaskID) == "" { continue } - _ = al.taskLedger.AddRemediation(finding.TaskID, tools.TaskRemediation{ - Action: "notify", - Status: "queued", - Note: finding.Message, + // Decide whether this finding should trigger an automatic rerun. + switch finding.Category { + case "missed": + // always eligible in retry modes + case "quality", "inconsistency": + if mode != "retry_all" { + continue + } + default: + continue + } + + entry, ok := al.taskLedger.Get(finding.TaskID) + if !ok { + continue + } + if retryLimit > 0 && entry.RetryCount >= retryLimit { + continue + } + + intent := strings.TrimSpace(entry.Intent) + if intent == "" { + _ = al.taskLedger.AddRemediation(entry.ID, tools.TaskRemediation{ + Action: "retry", + Status: "skipped", + Note: "missing task intent; cannot auto-retry", + }) + continue + } + + if hasRecentRetryRemediation(entry.Remediations, thresholdMS) { + continue + } + + originChannel, originChatID := al.resolveRemediationDestination(entry) + if originChannel == "" || originChatID == "" { + continue + } + + retryTask := buildRetryTask(finding, entry) + label := fmt.Sprintf("audit-%s:%s", finding.Category, entry.ID) + + taskInfo, err := defaultAgent.SubagentManager.SpawnTask( + ctx, + retryTask, + label, + targetAgentID, + originChannel, + originChatID, + nil, + ) + if err != nil { + _ = al.taskLedger.AddRemediation(entry.ID, tools.TaskRemediation{ + Action: "retry", + Status: "error", + Note: fmt.Sprintf("failed to spawn retry task: %v", err), + }) + continue + } + + _ = al.taskLedger.IncrementRetry(entry.ID) + _ = al.taskLedger.AddRemediation(entry.ID, tools.TaskRemediation{ + Action: "retry", + Status: "spawned", + Note: fmt.Sprintf("spawned %s (agent_id=%s)", taskInfo.ID, strings.TrimSpace(targetAgentID)), }) + spawned++ } } +func hasRecentRetryRemediation(remediations []tools.TaskRemediation, thresholdMS int64) bool { + for _, r := range remediations { + if strings.ToLower(strings.TrimSpace(r.Action)) != "retry" { + continue + } + if r.CreatedAtMS < thresholdMS { + continue + } + switch strings.ToLower(strings.TrimSpace(r.Status)) { + case "queued", "spawned", "running", "skipped": + return true + } + } + return false +} + +func (al *AgentLoop) resolveRemediationDestination(entry tools.TaskLedgerEntry) (string, string) { + channel := strings.TrimSpace(entry.OriginChannel) + chatID := strings.TrimSpace(entry.OriginChatID) + if channel != "" && chatID != "" && !constants.IsInternalChannel(channel) { + return channel, chatID + } + + channel, chatID = al.resolveAuditDestination() + if channel == "" || chatID == "" || constants.IsInternalChannel(channel) { + return "", "" + } + return channel, chatID +} + +func buildRetryTask(finding AuditFinding, entry tools.TaskLedgerEntry) string { + intent := strings.TrimSpace(entry.Intent) + if intent == "" { + return "" + } + + reason := strings.TrimSpace(finding.Message) + if reason == "" { + reason = "Task requires follow-up." + } + + var b strings.Builder + b.WriteString("You are running an automatic remediation retry for a previously problematic task.\n") + b.WriteString("Be concise and deliver a complete result.\n\n") + b.WriteString(fmt.Sprintf("Original task id: %s\n", entry.ID)) + b.WriteString(fmt.Sprintf("Original status: %s\n", entry.Status)) + b.WriteString(fmt.Sprintf("Finding category: %s\n", finding.Category)) + b.WriteString(fmt.Sprintf("Reason: %s\n\n", reason)) + b.WriteString("Task:\n") + b.WriteString(intent) + b.WriteString("\n\nAcceptance criteria:\n") + switch finding.Category { + case "quality": + b.WriteString("- Produce a non-empty result.\n- Include concrete deliverables.\n") + case "inconsistency": + b.WriteString("- If tools are required, use them and complete the task end-to-end.\n") + default: + b.WriteString("- Complete the task end-to-end.\n") + } + return b.String() +} + func (al *AgentLoop) publishAuditReport(report *AuditReport) { if report == nil || len(report.Findings) == 0 || al.bus == nil { return diff --git a/pkg/agent/audit_test.go b/pkg/agent/audit_test.go index a4ba55d66..dc078969b 100644 --- a/pkg/agent/audit_test.go +++ b/pkg/agent/audit_test.go @@ -87,3 +87,102 @@ func TestParseSupervisorReview_EmbeddedJSON(t *testing.T) { t.Fatalf("issue category = %q", review.Issues[0].Category) } } + +func TestApplyAutoRemediation_RetriesMissedTasksWithCooldown(t *testing.T) { + cfg := config.DefaultConfig() + cfg.Agents.Defaults.Workspace = t.TempDir() + cfg.Audit.Enabled = true + cfg.Audit.AutoRemediation = "retry_missed" + cfg.Audit.MaxAutoRemediationsPerCycle = 5 + cfg.Audit.RemediationCooldownMinutes = 60 + + loop := NewAgentLoop(cfg, bus.NewMessageBus(), &mockProvider{}) + ledger := loop.GetTaskLedger() + if ledger == nil { + t.Fatal("expected task ledger") + } + + taskID := "task-failed-1" + _ = ledger.UpsertTask(tools.TaskLedgerEntry{ + ID: taskID, + Status: tools.TaskStatusFailed, + Intent: "Write a short status update about the project.", + OriginChannel: "telegram", + OriginChatID: "chat-1", + CreatedAtMS: time.Now().Add(-10 * time.Minute).UnixMilli(), + }) + + report := &AuditReport{ + GeneratedAt: time.Now(), + Lookback: 180 * time.Minute, + TotalTasks: 1, + Findings: []AuditFinding{ + { + TaskID: taskID, + Category: "missed", + Severity: "medium", + Message: "Task failed and still has retry budget.", + Recommendation: "Retry this task automatically or manually.", + }, + }, + } + + loop.applyAutoRemediation(context.Background(), report) + + entry, ok := ledger.Get(taskID) + if !ok { + t.Fatal("expected task in ledger") + } + if entry.RetryCount != 1 { + t.Fatalf("RetryCount = %d, want %d", entry.RetryCount, 1) + } + hasSpawned := false + for _, r := range entry.Remediations { + if strings.EqualFold(r.Action, "retry") && strings.EqualFold(r.Status, "spawned") { + hasSpawned = true + break + } + } + if !hasSpawned { + t.Fatalf("expected spawned retry remediation, got: %+v", entry.Remediations) + } + + // Second run should respect cooldown and not spawn another retry. + loop.applyAutoRemediation(context.Background(), report) + entry2, _ := ledger.Get(taskID) + if entry2.RetryCount != 1 { + t.Fatalf("RetryCount after cooldown check = %d, want %d", entry2.RetryCount, 1) + } + spawnedCount := 0 + spawnedTaskID := "" + for _, r := range entry2.Remediations { + if strings.EqualFold(r.Action, "retry") && strings.EqualFold(r.Status, "spawned") { + spawnedCount++ + parts := strings.Fields(r.Note) + if len(parts) >= 2 { + spawnedTaskID = parts[1] + } + } + } + if spawnedCount != 1 { + t.Fatalf("spawned remediation count = %d, want %d", spawnedCount, 1) + } + + // Ensure the spawned subagent finishes before TempDir cleanup. + if spawnedTaskID == "" { + t.Fatal("expected spawned task id in remediation note") + } + deadline := time.Now().Add(2 * time.Second) + for { + subTask, ok := ledger.Get(spawnedTaskID) + if ok && (subTask.Status == tools.TaskStatusCompleted || + subTask.Status == tools.TaskStatusFailed || + subTask.Status == tools.TaskStatusCancelled) { + break + } + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for spawned task %s to finish", spawnedTaskID) + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/pkg/agent/instance.go b/pkg/agent/instance.go index d5fbdbc28..8c5073412 100644 --- a/pkg/agent/instance.go +++ b/pkg/agent/instance.go @@ -28,6 +28,7 @@ type AgentInstance struct { Sessions *session.SessionManager ContextBuilder *ContextBuilder Tools *tools.ToolRegistry + SubagentManager *tools.SubagentManager Subagents *config.SubagentsConfig SkillsFilter []string Candidates []providers.FallbackCandidate diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 9eaedae7e..5ca775b41 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -174,6 +174,7 @@ func registerSharedTools( cfg.Orchestration.ToolParallelOverrides, ) subagentManager.SetTools(agent.Tools) + agent.SubagentManager = subagentManager currentAgentID := agentID subagentManager.SetExecutionResolver(func(targetAgentID string) (tools.SubagentExecutionConfig, error) { return resolveSubagentExecution(cfg, registry, provider, currentAgentID, targetAgentID) diff --git a/pkg/config/config.go b/pkg/config/config.go index 4a15093f9..ed25c44e0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -377,7 +377,16 @@ type AuditConfig struct { MinConfidence float64 `json:"min_confidence" env:"PICOCLAW_AUDIT_MIN_CONFIDENCE"` InconsistencyPolicy string `json:"inconsistency_policy" env:"PICOCLAW_AUDIT_INCONSISTENCY_POLICY"` AutoRemediation string `json:"auto_remediation" env:"PICOCLAW_AUDIT_AUTO_REMEDIATION"` - NotifyChannel string `json:"notify_channel" env:"PICOCLAW_AUDIT_NOTIFY_CHANNEL"` + // MaxAutoRemediationsPerCycle limits how many retry/fix tasks can be spawned + // in one audit cycle to avoid runaway loops. + MaxAutoRemediationsPerCycle int `json:"max_auto_remediations_per_cycle" env:"PICOCLAW_AUDIT_MAX_AUTO_REMEDIATIONS_PER_CYCLE"` + // RemediationCooldownMinutes prevents re-triggering remediation for the same + // task too frequently. + RemediationCooldownMinutes int `json:"remediation_cooldown_minutes" env:"PICOCLAW_AUDIT_REMEDIATION_COOLDOWN_MINUTES"` + // RemediationAgentID optionally delegates remediation retries to a specific + // agent id (requires subagent allowlist when targeting a different agent). + RemediationAgentID string `json:"remediation_agent_id" env:"PICOCLAW_AUDIT_REMEDIATION_AGENT_ID"` + NotifyChannel string `json:"notify_channel" env:"PICOCLAW_AUDIT_NOTIFY_CHANNEL"` } type AuditSupervisorConfig struct { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index fa0885062..7336bb57b 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -364,13 +364,16 @@ func DefaultConfig() *Config { ToolParallelOverrides: map[string]string{}, }, Audit: AuditConfig{ - Enabled: false, - IntervalMinutes: 30, - LookbackMinutes: 180, - MinConfidence: 0.75, - InconsistencyPolicy: "strict", - AutoRemediation: "safe_only", - NotifyChannel: "last_active", + Enabled: false, + IntervalMinutes: 30, + LookbackMinutes: 180, + MinConfidence: 0.75, + InconsistencyPolicy: "strict", + AutoRemediation: "safe_only", + MaxAutoRemediationsPerCycle: 3, + RemediationCooldownMinutes: 10, + RemediationAgentID: "", + NotifyChannel: "last_active", Supervisor: AuditSupervisorConfig{ Enabled: false, Model: &AgentModelConfig{