You've already forked agentic-coding-workflow
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.
This commit is contained in:
+2
-2
@@ -254,7 +254,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
|
|||||||
Enabled: cfg.Notify.Webhook.Enabled,
|
Enabled: cfg.Notify.Webhook.Enabled,
|
||||||
Topics: cfg.Notify.Webhook.Topics,
|
Topics: cfg.Notify.Webhook.Topics,
|
||||||
Backoff: webhookBackoff,
|
Backoff: webhookBackoff,
|
||||||
}, clock.Real())
|
}, clock.Real(), auditRec, log)
|
||||||
notifyDispatcher.Register(webhookNotifier)
|
notifyDispatcher.Register(webhookNotifier)
|
||||||
|
|
||||||
dispAdapter := dispatcherAdapter{d: notifyDispatcher}
|
dispAdapter := dispatcherAdapter{d: notifyDispatcher}
|
||||||
@@ -269,7 +269,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
|
|||||||
runners.WorktreePruneConfig{
|
runners.WorktreePruneConfig{
|
||||||
IdleThreshold: cfg.Jobs.WorktreePrune.IdleThreshold,
|
IdleThreshold: cfg.Jobs.WorktreePrune.IdleThreshold,
|
||||||
WarningLead: cfg.Jobs.WorktreePrune.WarningLead,
|
WarningLead: cfg.Jobs.WorktreePrune.WarningLead,
|
||||||
}, log)
|
}, log, auditRec)
|
||||||
attCleanupRunner := runners.NewAttachmentCleanup(
|
attCleanupRunner := runners.NewAttachmentCleanup(
|
||||||
chatAttachmentAdapter{repo: chatRepo},
|
chatAttachmentAdapter{repo: chatRepo},
|
||||||
attachStorage,
|
attachStorage,
|
||||||
|
|||||||
@@ -556,7 +556,7 @@ func TestEndToEnd_WebhookClosedLoop(t *testing.T) {
|
|||||||
Enabled: true,
|
Enabled: true,
|
||||||
Topics: []string{"workspace.sync_failed"},
|
Topics: []string{"workspace.sync_failed"},
|
||||||
Backoff: []time.Duration{50 * time.Millisecond, 200 * time.Millisecond},
|
Backoff: []time.Duration{50 * time.Millisecond, 200 * time.Millisecond},
|
||||||
}, clock.Real())
|
}, clock.Real(), auditRec, log)
|
||||||
disp.Register(n)
|
disp.Register(n)
|
||||||
|
|
||||||
// Run a single Worker (no full Module needed for this test).
|
// 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() })
|
t.Cleanup(func() { wcancel(); w.Wait() })
|
||||||
go w.Run(wctx)
|
go w.Run(wctx)
|
||||||
|
|
||||||
// Suppress unused-variable vet warning.
|
|
||||||
_ = auditRec
|
|
||||||
|
|
||||||
eventTime := time.Now()
|
eventTime := time.Now()
|
||||||
require.NoError(t, disp.Dispatch(ctx, notify.Message{
|
require.NoError(t, disp.Dispatch(ctx, notify.Message{
|
||||||
ID: uuid.New(), UserID: uuid.New(),
|
ID: uuid.New(), UserID: uuid.New(),
|
||||||
|
|||||||
@@ -3,8 +3,10 @@ package jobs
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"log/slog"
|
||||||
"time"
|
"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/clock"
|
||||||
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
||||||
)
|
)
|
||||||
@@ -27,14 +29,20 @@ type WebhookNotifier struct {
|
|||||||
repo Repository
|
repo Repository
|
||||||
cfg WebhookNotifierConfig
|
cfg WebhookNotifierConfig
|
||||||
clk clock.Clock
|
clk clock.Clock
|
||||||
|
rec audit.Recorder
|
||||||
|
log *slog.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewWebhookNotifier constructs the notifier.
|
// NewWebhookNotifier constructs the notifier. rec is optional; when non-nil the
|
||||||
func NewWebhookNotifier(repo Repository, cfg WebhookNotifierConfig, clk clock.Clock) *WebhookNotifier {
|
// 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 {
|
if clk == nil {
|
||||||
clk = clock.Real()
|
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.
|
// 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 {
|
if maxAttempts < 1 {
|
||||||
maxAttempts = 1
|
maxAttempts = 1
|
||||||
}
|
}
|
||||||
_, err = n.repo.Enqueue(ctx, TypeWebhookDeliver, payload, n.clk.Now(), maxAttempts)
|
enqueued, err := n.repo.Enqueue(ctx, TypeWebhookDeliver, payload, n.clk.Now(), maxAttempts)
|
||||||
|
if err != nil {
|
||||||
return err
|
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 {
|
func topicAllowed(allow []string, t string) bool {
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ func TestWebhookNotifier_EnqueuesForWhitelistedTopic(t *testing.T) {
|
|||||||
Enabled: true,
|
Enabled: true,
|
||||||
Topics: []string{"workspace.sync_failed"},
|
Topics: []string{"workspace.sync_failed"},
|
||||||
Backoff: []time.Duration{30 * time.Second},
|
Backoff: []time.Duration{30 * time.Second},
|
||||||
}, clock.Real())
|
}, clock.Real(), nil, nil)
|
||||||
|
|
||||||
err := n.Send(ctx, notify.Message{
|
err := n.Send(ctx, notify.Message{
|
||||||
ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed",
|
ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed",
|
||||||
@@ -51,7 +51,7 @@ func TestWebhookNotifier_DropsNonWhitelistedTopic(t *testing.T) {
|
|||||||
Enabled: true,
|
Enabled: true,
|
||||||
Topics: []string{"workspace.sync_failed"},
|
Topics: []string{"workspace.sync_failed"},
|
||||||
Backoff: []time.Duration{30 * time.Second},
|
Backoff: []time.Duration{30 * time.Second},
|
||||||
}, clock.Real())
|
}, clock.Real(), nil, nil)
|
||||||
|
|
||||||
err := n.Send(ctx, notify.Message{
|
err := n.Send(ctx, notify.Message{
|
||||||
ID: uuid.New(), UserID: uuid.New(), Topic: "chat.message_failed",
|
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{
|
n := jobs.NewWebhookNotifier(repo, jobs.WebhookNotifierConfig{
|
||||||
Enabled: false,
|
Enabled: false,
|
||||||
Topics: []string{"workspace.sync_failed"},
|
Topics: []string{"workspace.sync_failed"},
|
||||||
}, clock.Real())
|
}, clock.Real(), nil, nil)
|
||||||
err := n.Send(ctx, notify.Message{
|
err := n.Send(ctx, notify.Message{
|
||||||
ID: uuid.New(), UserID: uuid.New(), Topic: "workspace.sync_failed", CreatedAt: time.Now(),
|
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) {
|
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())
|
assert.Equal(t, notify.ChannelWebhook, n.Name())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
|
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
|
|
||||||
|
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
||||||
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -50,14 +51,16 @@ type WorktreePrune struct {
|
|||||||
disp Dispatcher
|
disp Dispatcher
|
||||||
cfg WorktreePruneConfig
|
cfg WorktreePruneConfig
|
||||||
log *slog.Logger
|
log *slog.Logger
|
||||||
|
audit audit.Recorder
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewWorktreePrune constructs the runner.
|
// NewWorktreePrune constructs the runner. rec is optional; when non-nil the
|
||||||
func NewWorktreePrune(repo WorktreePruneRepo, gitr WorktreePruner, disp Dispatcher, cfg WorktreePruneConfig, log *slog.Logger) *WorktreePrune {
|
// 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 {
|
if log == nil {
|
||||||
log = slog.Default()
|
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.
|
// 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 {
|
if err := r.repo.FinishPruneSuccess(ctx, w.ID); err != nil {
|
||||||
r.log.Error("worktree_prune.finish_success", "worktree_id", w.ID, "err", err.Error())
|
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())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -95,7 +95,7 @@ func TestWorktreePrune_WarningPhase(t *testing.T) {
|
|||||||
gitr := &fakePruner{}
|
gitr := &fakePruner{}
|
||||||
cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour}
|
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())
|
r.Run(context.Background())
|
||||||
|
|
||||||
require.Len(t, repo.markedWarning, 1)
|
require.Len(t, repo.markedWarning, 1)
|
||||||
@@ -124,7 +124,7 @@ func TestWorktreePrune_ExecutionPhaseSuccess(t *testing.T) {
|
|||||||
gitr := &fakePruner{}
|
gitr := &fakePruner{}
|
||||||
cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour}
|
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())
|
r.Run(context.Background())
|
||||||
|
|
||||||
assert.True(t, repo.tryStartCalled[id])
|
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")}
|
gitr := &fakePruner{err: errors.New("git worktree remove failed")}
|
||||||
cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour}
|
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())
|
r.Run(context.Background())
|
||||||
|
|
||||||
assert.True(t, repo.tryStartCalled[id])
|
assert.True(t, repo.tryStartCalled[id])
|
||||||
@@ -162,7 +162,7 @@ func TestWorktreePrune_NoRowsIsNoOp(t *testing.T) {
|
|||||||
disp := &recordingDispatcher{}
|
disp := &recordingDispatcher{}
|
||||||
cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour}
|
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())
|
r.Run(context.Background())
|
||||||
|
|
||||||
assert.Empty(t, repo.markedWarning)
|
assert.Empty(t, repo.markedWarning)
|
||||||
@@ -185,7 +185,7 @@ func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) {
|
|||||||
gitr := &fakePruner{}
|
gitr := &fakePruner{}
|
||||||
cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour}
|
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())
|
r.Run(context.Background())
|
||||||
|
|
||||||
// TryStartPrune was called but failed.
|
// TryStartPrune was called but failed.
|
||||||
|
|||||||
Reference in New Issue
Block a user