From b1ccce567ec03108efd42e72e9aef80c042dd467 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 6 May 2026 13:35:45 +0800 Subject: [PATCH] feat(jobs): emit job.enqueue + worktree.pruned audit rows Deferred audit retrofits from Plan #5 self-review: - WebhookNotifier now records a job.enqueue audit row after each successful repo.Enqueue (best-effort: failures are warn-logged, not propagated). Added optional audit.Recorder + *slog.Logger params to NewWebhookNotifier. - WorktreePrune.executionPhase now records a worktree.pruned audit row after each successful FinishPruneSuccess. Added optional audit.Recorder param to NewWorktreePrune. - app.go wires the existing auditRec into both constructors. - integration_test.go: drop the _ = auditRec suppression and pass auditRec into the webhook notifier so the closed-loop test exercises the audit emit path. --- internal/app/app.go | 4 +-- internal/app/integration_test.go | 5 +-- internal/jobs/notifier.go | 34 +++++++++++++++++--- internal/jobs/notifier_test.go | 8 ++--- internal/jobs/runners/worktree_prune.go | 32 +++++++++++++----- internal/jobs/runners/worktree_prune_test.go | 10 +++--- 6 files changed, 65 insertions(+), 28 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index b100522..5473557 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -254,7 +254,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { Enabled: cfg.Notify.Webhook.Enabled, Topics: cfg.Notify.Webhook.Topics, Backoff: webhookBackoff, - }, clock.Real()) + }, clock.Real(), auditRec, log) notifyDispatcher.Register(webhookNotifier) dispAdapter := dispatcherAdapter{d: notifyDispatcher} @@ -269,7 +269,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { runners.WorktreePruneConfig{ IdleThreshold: cfg.Jobs.WorktreePrune.IdleThreshold, WarningLead: cfg.Jobs.WorktreePrune.WarningLead, - }, log) + }, log, auditRec) attCleanupRunner := runners.NewAttachmentCleanup( chatAttachmentAdapter{repo: chatRepo}, attachStorage, diff --git a/internal/app/integration_test.go b/internal/app/integration_test.go index e181f75..07e2210 100644 --- a/internal/app/integration_test.go +++ b/internal/app/integration_test.go @@ -556,7 +556,7 @@ func TestEndToEnd_WebhookClosedLoop(t *testing.T) { Enabled: true, Topics: []string{"workspace.sync_failed"}, Backoff: []time.Duration{50 * time.Millisecond, 200 * time.Millisecond}, - }, clock.Real()) + }, clock.Real(), auditRec, log) disp.Register(n) // Run a single Worker (no full Module needed for this test). @@ -569,9 +569,6 @@ func TestEndToEnd_WebhookClosedLoop(t *testing.T) { t.Cleanup(func() { wcancel(); w.Wait() }) go w.Run(wctx) - // Suppress unused-variable vet warning. - _ = auditRec - eventTime := time.Now() require.NoError(t, disp.Dispatch(ctx, notify.Message{ ID: uuid.New(), UserID: uuid.New(), diff --git a/internal/jobs/notifier.go b/internal/jobs/notifier.go index 13395d2..75775a9 100644 --- a/internal/jobs/notifier.go +++ b/internal/jobs/notifier.go @@ -3,8 +3,10 @@ package jobs import ( "context" "encoding/json" + "log/slog" "time" + "github.com/yan1h/agent-coding-workflow/internal/audit" "github.com/yan1h/agent-coding-workflow/internal/infra/clock" "github.com/yan1h/agent-coding-workflow/internal/infra/notify" ) @@ -27,14 +29,20 @@ type WebhookNotifier struct { repo Repository cfg WebhookNotifierConfig clk clock.Clock + rec audit.Recorder + log *slog.Logger } -// NewWebhookNotifier constructs the notifier. -func NewWebhookNotifier(repo Repository, cfg WebhookNotifierConfig, clk clock.Clock) *WebhookNotifier { +// NewWebhookNotifier constructs the notifier. rec is optional; when non-nil the +// notifier emits a job.enqueue audit row after each successful Enqueue. +func NewWebhookNotifier(repo Repository, cfg WebhookNotifierConfig, clk clock.Clock, rec audit.Recorder, log *slog.Logger) *WebhookNotifier { if clk == nil { clk = clock.Real() } - return &WebhookNotifier{repo: repo, cfg: cfg, clk: clk} + if log == nil { + log = slog.Default() + } + return &WebhookNotifier{repo: repo, cfg: cfg, clk: clk, rec: rec, log: log} } // Name reports the notify.Channel for this notifier. @@ -66,8 +74,24 @@ func (n *WebhookNotifier) Send(ctx context.Context, msg notify.Message) error { if maxAttempts < 1 { maxAttempts = 1 } - _, err = n.repo.Enqueue(ctx, TypeWebhookDeliver, payload, n.clk.Now(), maxAttempts) - return err + enqueued, err := n.repo.Enqueue(ctx, TypeWebhookDeliver, payload, n.clk.Now(), maxAttempts) + if err != nil { + return err + } + if n.rec != nil { + if rerr := n.rec.Record(ctx, audit.Entry{ + Action: "job.enqueue", + TargetType: "job", + TargetID: enqueued.ID.String(), + Metadata: map[string]any{ + "topic": msg.Topic, + "job_type": string(TypeWebhookDeliver), + }, + }); rerr != nil { + n.log.Warn("webhook_notifier.audit_failed", "job_id", enqueued.ID, "err", rerr.Error()) + } + } + return nil } func topicAllowed(allow []string, t string) bool { diff --git a/internal/jobs/notifier_test.go b/internal/jobs/notifier_test.go index f9233a6..4d71ac1 100644 --- a/internal/jobs/notifier_test.go +++ b/internal/jobs/notifier_test.go @@ -25,7 +25,7 @@ func TestWebhookNotifier_EnqueuesForWhitelistedTopic(t *testing.T) { Enabled: true, Topics: []string{"workspace.sync_failed"}, Backoff: []time.Duration{30 * time.Second}, - }, clock.Real()) + }, clock.Real(), nil, nil) err := n.Send(ctx, notify.Message{ ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed", @@ -51,7 +51,7 @@ func TestWebhookNotifier_DropsNonWhitelistedTopic(t *testing.T) { Enabled: true, Topics: []string{"workspace.sync_failed"}, Backoff: []time.Duration{30 * time.Second}, - }, clock.Real()) + }, clock.Real(), nil, nil) err := n.Send(ctx, notify.Message{ ID: uuid.New(), UserID: uuid.New(), Topic: "chat.message_failed", @@ -69,7 +69,7 @@ func TestWebhookNotifier_DisabledShortCircuits(t *testing.T) { n := jobs.NewWebhookNotifier(repo, jobs.WebhookNotifierConfig{ Enabled: false, Topics: []string{"workspace.sync_failed"}, - }, clock.Real()) + }, clock.Real(), nil, nil) err := n.Send(ctx, notify.Message{ ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed", CreatedAt: time.Now(), }) @@ -79,6 +79,6 @@ func TestWebhookNotifier_DisabledShortCircuits(t *testing.T) { } func TestWebhookNotifier_Name(t *testing.T) { - n := jobs.NewWebhookNotifier(nil, jobs.WebhookNotifierConfig{}, clock.Real()) + n := jobs.NewWebhookNotifier(nil, jobs.WebhookNotifierConfig{}, clock.Real(), nil, nil) assert.Equal(t, notify.ChannelWebhook, n.Name()) } diff --git a/internal/jobs/runners/worktree_prune.go b/internal/jobs/runners/worktree_prune.go index 41826b7..8222739 100644 --- a/internal/jobs/runners/worktree_prune.go +++ b/internal/jobs/runners/worktree_prune.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" + "github.com/yan1h/agent-coding-workflow/internal/audit" "github.com/yan1h/agent-coding-workflow/internal/infra/notify" ) @@ -45,19 +46,21 @@ type WorktreePruneConfig struct { // WorktreePrune is the B runner. type WorktreePrune struct { - repo WorktreePruneRepo - gitr WorktreePruner - disp Dispatcher - cfg WorktreePruneConfig - log *slog.Logger + repo WorktreePruneRepo + gitr WorktreePruner + disp Dispatcher + cfg WorktreePruneConfig + log *slog.Logger + audit audit.Recorder } -// NewWorktreePrune constructs the runner. -func NewWorktreePrune(repo WorktreePruneRepo, gitr WorktreePruner, disp Dispatcher, cfg WorktreePruneConfig, log *slog.Logger) *WorktreePrune { +// NewWorktreePrune constructs the runner. rec is optional; when non-nil the +// runner emits a worktree.pruned audit row after each successful prune. +func NewWorktreePrune(repo WorktreePruneRepo, gitr WorktreePruner, disp Dispatcher, cfg WorktreePruneConfig, log *slog.Logger, rec audit.Recorder) *WorktreePrune { if log == nil { log = slog.Default() } - return &WorktreePrune{repo: repo, gitr: gitr, disp: disp, cfg: cfg, log: log} + return &WorktreePrune{repo: repo, gitr: gitr, disp: disp, cfg: cfg, log: log, audit: rec} } // Run executes warning phase then execution phase. @@ -117,6 +120,19 @@ func (r *WorktreePrune) executionPhase(ctx context.Context) { } if err := r.repo.FinishPruneSuccess(ctx, w.ID); err != nil { r.log.Error("worktree_prune.finish_success", "worktree_id", w.ID, "err", err.Error()) + continue + } + if r.audit != nil { + if rerr := r.audit.Record(ctx, audit.Entry{ + Action: "worktree.pruned", + TargetType: "worktree", + TargetID: w.ID.String(), + Metadata: map[string]any{ + "branch": w.Branch, + }, + }); rerr != nil { + r.log.Warn("worktree_prune.audit_failed", "worktree_id", w.ID, "err", rerr.Error()) + } } } } diff --git a/internal/jobs/runners/worktree_prune_test.go b/internal/jobs/runners/worktree_prune_test.go index 2ba30f4..3243c04 100644 --- a/internal/jobs/runners/worktree_prune_test.go +++ b/internal/jobs/runners/worktree_prune_test.go @@ -95,7 +95,7 @@ func TestWorktreePrune_WarningPhase(t *testing.T) { gitr := &fakePruner{} cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} - r := runners.NewWorktreePrune(repo, gitr, disp, cfg, discardLogger()) + r := runners.NewWorktreePrune(repo, gitr, disp, cfg, discardLogger(), nil) r.Run(context.Background()) require.Len(t, repo.markedWarning, 1) @@ -124,7 +124,7 @@ func TestWorktreePrune_ExecutionPhaseSuccess(t *testing.T) { gitr := &fakePruner{} cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} - r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger()) + r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger(), nil) r.Run(context.Background()) assert.True(t, repo.tryStartCalled[id]) @@ -146,7 +146,7 @@ func TestWorktreePrune_ExecutionFailureRollsBack(t *testing.T) { gitr := &fakePruner{err: errors.New("git worktree remove failed")} cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} - r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger()) + r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger(), nil) r.Run(context.Background()) assert.True(t, repo.tryStartCalled[id]) @@ -162,7 +162,7 @@ func TestWorktreePrune_NoRowsIsNoOp(t *testing.T) { disp := &recordingDispatcher{} cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} - r := runners.NewWorktreePrune(repo, gitr, disp, cfg, discardLogger()) + r := runners.NewWorktreePrune(repo, gitr, disp, cfg, discardLogger(), nil) r.Run(context.Background()) assert.Empty(t, repo.markedWarning) @@ -185,7 +185,7 @@ func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) { gitr := &fakePruner{} cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} - r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger()) + r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger(), nil) r.Run(context.Background()) // TryStartPrune was called but failed.