feat(intercept): 增加审批记录详情与执行审计

This commit is contained in:
ji ruo committed 2026-09-10 15:10:50 +08:00
1 parent 8dae851b9b
commit 9ad323dc4c
22 files changed
+1687 -535

No files matched your search

+42 -15
View File
@@ -3,6 +3,7 @@ package db
import (
"database/sql"
"encoding/json"
"strings"
"time"
)
@@ -34,6 +35,7 @@ type InterceptPending struct {
ToolName string `json:"tool_name"`
ToolInput json.RawMessage `json:"tool_input"`
Status string `json:"status"`
DecisionSource string `json:"decision_source"`
Reason string `json:"reason"` // 规则 message 或模型判定理由(前缀 [模型])
DecidedAt *time.Time `json:"decided_at"`
CreatedAt time.Time `json:"created_at"`
@@ -106,7 +108,7 @@ func (d *DB) ToggleInterceptRule(id int64, enabled bool) error {
// CreateInterceptPending inserts a pending approval record and returns its ID.
// convID == 0 → conversation_id stored as NULL (background task).
// taskID == "" → task_id stored as NULL.
func (d *DB) CreateInterceptPending(ruleID, convID int64, taskID, agentName, toolName string, input []byte, reason string) (int64, error) {
func (d *DB) CreateInterceptPending(ruleID, convID int64, taskID, agentName, toolName string, input []byte, reason string, audits ...*InterceptAudit) (int64, error) {
raw := json.RawMessage(input)
if len(raw) == 0 {
raw = json.RawMessage("{}")
@@ -126,9 +128,9 @@ func (d *DB) CreateInterceptPending(ruleID, convID int64, taskID, agentName, too
}
var id int64
err := d.QueryRow(`
INSERT INTO intercept_pending(rule_id, conversation_id, task_id, agent_name, tool_name, tool_input, reason)
VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id`,
ruleIDPtr, convIDPtr, taskIDPtr, agentName, toolName, raw, reason).Scan(&id)
INSERT INTO intercept_pending(rule_id, conversation_id, task_id, agent_name, tool_name, tool_input, reason, decision_source, audit)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING id`,
ruleIDPtr, convIDPtr, taskIDPtr, agentName, toolName, raw, reason, interceptSource(ruleID, reason), firstAudit(audits)).Scan(&id)
return id, err
}
@@ -142,7 +144,7 @@ func (d *DB) DecideInterceptPending(id int64, status string) error {
// (status = 'allowed' | 'denied'), decided_at stamped now. Used to log allow/deny
// rule matches for observability — they don't block and need no user action, so unlike
// CreateInterceptPending (which starts 'pending') this records the outcome directly.
func (d *DB) CreateDecidedIntercept(ruleID, convID int64, taskID, agentName, toolName string, input []byte, status, reason string) (int64, error) {
func (d *DB) CreateDecidedIntercept(ruleID, convID int64, taskID, agentName, toolName string, input []byte, status, reason string, audits ...*InterceptAudit) (int64, error) {
raw := json.RawMessage(input)
if len(raw) == 0 {
raw = json.RawMessage("{}")
@@ -162,17 +164,17 @@ func (d *DB) CreateDecidedIntercept(ruleID, convID int64, taskID, agentName, too
}
var id int64
err := d.QueryRow(`
INSERT INTO intercept_pending(rule_id, conversation_id, task_id, agent_name, tool_name, tool_input, status, reason, decided_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW()) RETURNING id`,
ruleIDPtr, convIDPtr, taskIDPtr, agentName, toolName, raw, status, reason).Scan(&id)
INSERT INTO intercept_pending(rule_id, conversation_id, task_id, agent_name, tool_name, tool_input, status, reason, decided_at, decision_source, audit)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), $9, $10) RETURNING id`,
ruleIDPtr, convIDPtr, taskIDPtr, agentName, toolName, raw, status, reason, interceptSource(ruleID, reason), firstAudit(audits)).Scan(&id)
return id, err
}
const interceptPendingCols = `id, rule_id, conversation_id, task_id, agent_name, tool_name, tool_input, status, reason, decided_at, created_at`
const interceptPendingCols = `id, rule_id, conversation_id, task_id, agent_name, tool_name, tool_input, status, reason, decided_at, created_at, decision_source`
func scanInterceptPending(s interface{ Scan(...any) error }, p *InterceptPending) error {
return s.Scan(&p.ID, &p.RuleID, &p.ConversationID, &p.TaskID, &p.AgentName,
&p.ToolName, &p.ToolInput, &p.Status, &p.Reason, &p.DecidedAt, &p.CreatedAt)
&p.ToolName, &p.ToolInput, &p.Status, &p.Reason, &p.DecidedAt, &p.CreatedAt, &p.DecisionSource)
}
// ListPendingIntercepts returns all unresolved approval requests, newest first.
@@ -218,20 +220,45 @@ func scanInterceptApprovalRow(rows interface{ Scan(...any) error }, r *Intercept
return rows.Scan(
&r.ID, &r.RuleID, &r.ConversationID, &r.TaskID, &r.AgentName,
&r.ToolName, &r.ToolInput, &r.Status, &r.Reason, &r.DecidedAt, &r.CreatedAt,
&r.ConvTitle, &r.ConvAgentKey, &r.RuleName,
&r.DecisionSource, &r.ConvTitle, &r.ConvAgentKey, &r.RuleName,
)
}
const approvalRowSelect = `
SELECT ip.id, ip.rule_id, ip.conversation_id, ip.task_id, ip.agent_name,
ip.tool_name, ip.tool_input, ip.status, ip.reason, ip.decided_at, ip.created_at,
const approvalRowColumns = `ip.id, ip.rule_id, ip.conversation_id, ip.task_id, ip.agent_name,
ip.tool_name, ip.tool_input, ip.status, ip.reason, ip.decided_at, ip.created_at, ip.decision_source,
COALESCE(c.title,'') AS conv_title,
COALESCE(c.agent_key,'') AS conv_agent_key,
COALESCE(ir.name,'') AS rule_name
COALESCE(ir.name,'') AS rule_name`
const approvalRowJoins = `
FROM intercept_pending ip
LEFT JOIN conversations c ON c.id = ip.conversation_id
LEFT JOIN intercept_rules ir ON ir.id = ip.rule_id`
const approvalRowSelect = `SELECT ` + approvalRowColumns + approvalRowJoins
const approvalRowSelectWithAudit = `SELECT ` + approvalRowColumns + `, ip.audit` + approvalRowJoins
func interceptSource(ruleID int64, reason string) string {
if ruleID != 0 {
return "rule"
}
if strings.HasPrefix(reason, "[模型]") {
return "model"
}
return "unknown"
}
func firstAudit(audits []*InterceptAudit) any {
if len(audits) == 0 || audits[0] == nil {
return nil
}
raw, err := json.Marshal(audits[0])
if err != nil {
return nil
}
return raw
}
// ListAllIntercepts returns up to limit intercept_pending rows (newest first)
// joined with conversation and rule info.
func (d *DB) ListAllIntercepts(limit int) ([]InterceptApprovalRow, error) {
+112
View File
@@ -0,0 +1,112 @@
package db
import (
"database/sql"
"encoding/json"
"time"
)
// InterceptContextEntry is a bounded, recorded session event, not model reasoning.
type InterceptContextEntry struct {
Kind string `json:"kind"`
Tool string `json:"tool,omitempty"`
ToolUseID string `json:"tool_use_id,omitempty"`
Text string `json:"text"`
IsError bool `json:"is_error,omitempty"`
Truncated bool `json:"truncated,omitempty"`
}
// InterceptAudit is captured at review time. It is deliberately excluded from
// polling/list responses; old rows have no audit instead of reconstructed data.
type InterceptAudit struct {
RunID string `json:"run_id,omitempty"`
ToolUseID string `json:"tool_use_id,omitempty"`
Correlation string `json:"correlation"` // exact | ambiguous | unavailable
InputDigest string `json:"input_digest"`
UserMessage string `json:"user_message"`
UserTruncated bool `json:"user_truncated,omitempty"`
Context []InterceptContextEntry `json:"context"`
ContextTruncated bool `json:"context_truncated,omitempty"`
CapturedAt time.Time `json:"captured_at"`
ModelFallback bool `json:"model_fallback,omitempty"`
InitialAction string `json:"initial_action"`
InitialReason string `json:"initial_reason"`
EffectiveAction string `json:"effective_action,omitempty"`
DecisionReason string `json:"decision_reason,omitempty"`
RuleName string `json:"rule_name,omitempty"`
ConfigDigest string `json:"config_digest,omitempty"`
ProfileID int64 `json:"profile_id,omitempty"`
ExecutionStatus string `json:"execution_status"`
Output string `json:"output,omitempty"`
OutputTruncated bool `json:"output_truncated,omitempty"`
ExecutionEndedAt *time.Time `json:"execution_ended_at,omitempty"`
}
type InterceptDetail struct {
InterceptApprovalRow
Audit *InterceptAudit `json:"audit"`
}
func (d *DB) GetInterceptDetail(id int64) (*InterceptDetail, error) {
var out InterceptDetail
var raw []byte
err := d.QueryRow(approvalRowSelectWithAudit+` WHERE ip.id=$1`, id).Scan(
&out.ID, &out.RuleID, &out.ConversationID, &out.TaskID, &out.AgentName,
&out.ToolName, &out.ToolInput, &out.Status, &out.Reason, &out.DecidedAt, &out.CreatedAt,
&out.DecisionSource, &out.ConvTitle, &out.ConvAgentKey, &out.RuleName, &raw,
)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
if len(raw) > 0 {
if err := json.Unmarshal(raw, &out.Audit); err != nil {
return nil, err
}
}
return &out, nil
}
// ResolveIntercept atomically settles a pending request. A timeout cannot
// overwrite a human decision and repeat decisions cannot rewrite history.
func (d *DB) ResolveIntercept(id int64, status, action, reason string) (bool, error) {
execution := "not_executed"
if action == "allow" {
execution = "awaiting_result"
}
patch, err := json.Marshal(map[string]any{
"effective_action": action, "decision_reason": reason, "execution_status": execution,
})
if err != nil {
return false, err
}
r, err := d.Exec(`UPDATE intercept_pending SET status=$2, decided_at=NOW(),
audit=CASE WHEN audit IS NULL THEN NULL ELSE audit || $3::jsonb ||
CASE WHEN $4='allow' AND audit->>'correlation' IS DISTINCT FROM 'exact'
THEN '{"execution_status":"unknown"}'::jsonb ELSE '{}'::jsonb END END
WHERE id=$1 AND status='pending'`, id, status, patch, action)
if err != nil {
return false, err
}
n, err := r.RowsAffected()
return n == 1, err
}
// CompleteIntercept only updates the exact recorded call after it was allowed.
// A blocked tool_result must never be presented as a failed execution.
func (d *DB) CompleteIntercept(id int64, runID, toolUseID, status, output string, truncated bool) error {
patch, err := json.Marshal(map[string]any{
"execution_status": status, "output": output, "output_truncated": truncated,
"execution_ended_at": time.Now().UTC(),
})
if err != nil {
return err
}
_, err = d.Exec(`UPDATE intercept_pending SET audit=audit || $4::jsonb
WHERE id=$1 AND audit->>'run_id'=$2 AND audit->>'tool_use_id'=$3
AND audit->>'effective_action'='allow' AND audit->>'execution_status'='awaiting_result'`,
id, runID, toolUseID, patch)
return err
}
+167
View File
@@ -0,0 +1,167 @@
package db
import (
"encoding/json"
"strings"
"sync"
"sync/atomic"
"testing"
)
func TestInterceptDetails(t *testing.T) {
d, err := Open(testDSN(t))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = d.Close() })
create := func(t *testing.T, audit *InterceptAudit) int64 {
t.Helper()
id, err := d.CreateInterceptPending(0, 0, "approval-detail-test", "test", "Write", []byte(`{"path":"report.md"}`), "[模型] 请确认", audit)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _, _ = d.Exec(`DELETE FROM intercept_pending WHERE id=$1`, id) })
return id
}
t.Run("legacy and lazy payload", func(t *testing.T) {
id := create(t, nil)
got, err := d.GetInterceptDetail(id)
if err != nil || got == nil || got.Audit != nil {
t.Fatalf("legacy: %+v %v", got, err)
}
create(t, &InterceptAudit{UserMessage: "snapshot-only-marker", InitialAction: "ask"})
items, err := d.ListTaskIntercepts("approval-detail-test")
if err != nil {
t.Fatal(err)
}
raw, _ := json.Marshal(items)
if strings.Contains(string(raw), "snapshot-only-marker") || strings.Contains(string(raw), `"audit"`) {
t.Fatal("snapshot leaked into list response")
}
missing, err := d.GetInterceptDetail(-1)
if err != nil || missing != nil {
t.Fatal("missing record not reported")
}
})
t.Run("decision race and exact output", func(t *testing.T) {
id := create(t, &InterceptAudit{RunID: "run", ToolUseID: "call", Correlation: "exact", InitialAction: "ask", ExecutionStatus: "not_started"})
var wins atomic.Int32
var wg sync.WaitGroup
for range 8 {
wg.Go(func() {
ok, err := d.ResolveIntercept(id, "allowed", "allow", "人工允许执行")
if err != nil {
t.Error(err)
}
if ok {
wins.Add(1)
}
})
}
wg.Wait()
if wins.Load() != 1 {
t.Fatalf("%d decisions won", wins.Load())
}
ok, err := d.ResolveIntercept(id, "timeout", "deny", "late timeout")
if err != nil || ok {
t.Fatal("timeout overwrote decision")
}
if err := d.CompleteIntercept(id, "different-run", "call", "succeeded", "WRONG", false); err != nil {
t.Fatal(err)
}
got, _ := d.GetInterceptDetail(id)
if got.Audit.Output != "" {
t.Fatal("cross-run result attached")
}
if err := d.CompleteIntercept(id, "run", "call", "failed", "permission denied", true); err != nil {
t.Fatal(err)
}
got, err = d.GetInterceptDetail(id)
if err != nil || got.Status != "allowed" || got.Audit.InitialAction != "ask" || got.Audit.ExecutionStatus != "failed" || !got.Audit.OutputTruncated {
t.Fatalf("wrong details: %+v %v", got, err)
}
if err := d.CompleteIntercept(id, "run", "call", "succeeded", "late duplicate", false); err != nil {
t.Fatal(err)
}
got, _ = d.GetInterceptDetail(id)
if got.Audit.Output != "permission denied" {
t.Fatal("duplicate result rewrote output")
}
})
t.Run("denied output and timeout allow", func(t *testing.T) {
id := create(t, &InterceptAudit{RunID: "run", ToolUseID: "call", ExecutionStatus: "not_started"})
if _, err := d.ResolveIntercept(id, "denied", "deny", "人工拒绝"); err != nil {
t.Fatal(err)
}
if err := d.CompleteIntercept(id, "run", "call", "failed", "Blocked by hook", false); err != nil {
t.Fatal(err)
}
got, _ := d.GetInterceptDetail(id)
if got.Audit.ExecutionStatus != "not_executed" || got.Audit.Output != "" {
t.Fatal("denial presented as executed")
}
id = create(t, &InterceptAudit{RunID: "run2", ToolUseID: "call2", Correlation: "exact", InitialAction: "ask"})
if _, err := d.ResolveIntercept(id, "timeout", "allow", "超时允许"); err != nil {
t.Fatal(err)
}
if err := d.CompleteIntercept(id, "run2", "call2", "succeeded", "ok", false); err != nil {
t.Fatal(err)
}
got, _ = d.GetInterceptDetail(id)
if got.Status != "timeout" || got.Audit.EffectiveAction != "allow" || got.Audit.ExecutionStatus != "succeeded" {
t.Fatal("timeout action lost")
}
})
t.Run("archive compatibility", func(t *testing.T) {
for _, legacy := range []bool{true, false} {
id := create(t, &InterceptAudit{InitialAction: "ask", ExecutionStatus: "not_started"})
var raw []byte
if err := d.QueryRow(`SELECT row_to_json(ip) FROM intercept_pending ip WHERE id=$1`, id).Scan(&raw); err != nil {
t.Fatal(err)
}
var row map[string]any
if err := json.Unmarshal(raw, &row); err != nil {
t.Fatal(err)
}
if legacy {
delete(row, "audit")
delete(row, "decision_source")
}
archived, _ := json.Marshal([]map[string]any{row})
tx, err := d.Begin()
if err != nil {
t.Fatal(err)
}
defer tx.Rollback()
if _, err := tx.Exec(`DELETE FROM intercept_pending WHERE id=$1`, id); err != nil {
t.Fatal(err)
}
if err := restoreInterceptRows(tx, archived); err != nil {
t.Fatalf("legacy=%t: %v", legacy, err)
}
var status, source string
var auditJSON []byte
if err := tx.QueryRow(`SELECT status, decision_source, audit FROM intercept_pending WHERE id=$1`, id).Scan(&status, &source, &auditJSON); err != nil {
t.Fatal(err)
}
if status != "timeout" || source != "model" {
t.Fatalf("restored %s/%s", status, source)
}
if legacy && len(auditJSON) > 0 {
t.Fatal("fabricated legacy audit")
}
if !legacy {
var a InterceptAudit
if err := json.Unmarshal(auditJSON, &a); err != nil {
t.Fatal(err)
}
if a.InitialAction != "ask" || a.EffectiveAction != "deny" || a.ExecutionStatus != "not_executed" {
t.Fatalf("restored audit: %+v", a)
}
}
if err := tx.Rollback(); err != nil {
t.Fatal(err)
}
}
})
}
+5
View File
@@ -1024,6 +1024,11 @@ CREATE INDEX IF NOT EXISTS idx_intercept_pending_status ON intercept_pending(sta
CREATE INDEX IF NOT EXISTS idx_intercept_pending_task ON intercept_pending(task_id, created_at DESC);
-- 补旧库:reason 列(已发版,加列要带 IF NOT EXISTS)。
ALTER TABLE intercept_pending ADD COLUMN IF NOT EXISTS reason TEXT NOT NULL DEFAULT '';
-- Detail payloads are lazy-loaded; NULL preserves the meaning of legacy history.
ALTER TABLE intercept_pending ADD COLUMN IF NOT EXISTS audit JSONB;
ALTER TABLE intercept_pending ADD COLUMN IF NOT EXISTS decision_source TEXT NOT NULL DEFAULT '';
UPDATE intercept_pending SET decision_source=CASE WHEN rule_id IS NOT NULL THEN 'rule'
WHEN reason LIKE '[模型]%' THEN 'model' ELSE 'unknown' END WHERE decision_source='';
-- =====================================================================
-- L. 漏洞发现持久化
+18
View File
@@ -634,10 +634,28 @@ func restoreInterceptRows(tx *sql.Tx, raw json.RawMessage) error {
return err
}
for _, row := range rows {
// Archives predating approval snapshots lack this NOT NULL column.
if source, _ := row["decision_source"].(string); source == "" {
source = "unknown"
if row["rule_id"] != nil {
source = "rule"
} else if reason, _ := row["reason"].(string); strings.HasPrefix(reason, "[模型]") {
source = "model"
}
row["decision_source"] = source
}
if status, _ := row["status"].(string); status == "pending" {
row["status"] = "timeout"
row["reason"] = "任务归档期间审批已超时"
row["decided_at"] = time.Now().UTC()
if audit, ok := row["audit"].(map[string]any); ok {
audit["effective_action"] = "deny"
audit["decision_reason"] = "任务归档期间审批已超时"
audit["execution_status"] = "not_executed"
}
} else if audit, ok := row["audit"].(map[string]any); ok && audit["execution_status"] == "awaiting_result" {
// An archived run cannot resume its former result callback.
audit["execution_status"] = "unknown"
}
}
if len(rows) == 0 {