feat(T3.4): implement human-in-the-loop approval gates
- Add internal/approval package for workflow approval gates - Implement ApprovalGate for gating workflow progression - Implement ApprovalGateManager for managing multiple gates - Gate status tracking: pending, approved, rejected, expired - TTL-based gate expiration (auto-expire after timeout) - Multiple approval tracking (configurable approval count) - History tracking for all approval decisions - Query by task, workflow, status - Audit trail with decision reasons - 16 approval tests, all passing
This commit is contained in:
Test
@@ -0,0 +1,342 @@
|
||||
package approval
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ApprovalStatus represents approval state
|
||||
type ApprovalStatus string
|
||||
|
||||
const (
|
||||
StatusPending ApprovalStatus = "pending"
|
||||
StatusApproved ApprovalStatus = "approved"
|
||||
StatusRejected ApprovalStatus = "rejected"
|
||||
StatusExpired ApprovalStatus = "expired"
|
||||
)
|
||||
|
||||
// ApprovalDecision represents an approval decision
|
||||
type ApprovalDecision struct {
|
||||
Status ApprovalStatus `json:"status"`
|
||||
ApprovedBy string `json:"approved_by"`
|
||||
RejectedBy string `json:"rejected_by"`
|
||||
Reason string `json:"reason"`
|
||||
Comments string `json:"comments"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
|
||||
// ApprovalGate represents a human approval gate
|
||||
type ApprovalGate struct {
|
||||
ID string
|
||||
TaskID string
|
||||
WorkflowID string
|
||||
Description string
|
||||
Decision *ApprovalDecision
|
||||
CreatedAt time.Time
|
||||
TTL time.Duration // Time until gate expires
|
||||
RequiredApprovals int // Number of approvals needed (1 or more)
|
||||
Approvals []string // List of approvers
|
||||
}
|
||||
|
||||
// ApprovalGateManager manages approval gates
|
||||
type ApprovalGateManager struct {
|
||||
mu sync.RWMutex
|
||||
gates map[string]*ApprovalGate
|
||||
decisions map[string]*ApprovalDecision
|
||||
history []*ApprovalRecord
|
||||
stats *ApprovalStats
|
||||
}
|
||||
|
||||
// ApprovalRecord tracks approval history
|
||||
type ApprovalRecord struct {
|
||||
GateID string
|
||||
Decision ApprovalStatus
|
||||
ApprovedBy string
|
||||
RejectedBy string
|
||||
Timestamp time.Time
|
||||
Reason string
|
||||
}
|
||||
|
||||
// ApprovalStats tracks approval statistics
|
||||
type ApprovalStats struct {
|
||||
TotalGates int
|
||||
ApprovedGates int
|
||||
RejectedGates int
|
||||
PendingGates int
|
||||
ExpiredGates int
|
||||
AverageWaitTime time.Duration
|
||||
}
|
||||
|
||||
// NewApprovalGateManager creates a new approval gate manager
|
||||
func NewApprovalGateManager() *ApprovalGateManager {
|
||||
return &ApprovalGateManager{
|
||||
gates: make(map[string]*ApprovalGate),
|
||||
decisions: make(map[string]*ApprovalDecision),
|
||||
history: make([]*ApprovalRecord, 0),
|
||||
stats: &ApprovalStats{
|
||||
AverageWaitTime: 0,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// CreateGate creates a new approval gate
|
||||
func (agm *ApprovalGateManager) CreateGate(taskID, workflowID, description string, ttl time.Duration) *ApprovalGate {
|
||||
if ttl == 0 {
|
||||
ttl = 24 * time.Hour // Default 24 hours
|
||||
}
|
||||
|
||||
gate := &ApprovalGate{
|
||||
ID: fmt.Sprintf("gate-%d", time.Now().UnixNano()),
|
||||
TaskID: taskID,
|
||||
WorkflowID: workflowID,
|
||||
Description: description,
|
||||
CreatedAt: time.Now(),
|
||||
TTL: ttl,
|
||||
RequiredApprovals: 1,
|
||||
Approvals: make([]string, 0),
|
||||
Decision: &ApprovalDecision{
|
||||
Status: StatusPending,
|
||||
ExpiresAt: time.Now().Add(ttl),
|
||||
},
|
||||
}
|
||||
|
||||
agm.mu.Lock()
|
||||
defer agm.mu.Unlock()
|
||||
|
||||
agm.gates[gate.ID] = gate
|
||||
agm.decisions[gate.ID] = gate.Decision
|
||||
agm.stats.TotalGates++
|
||||
agm.stats.PendingGates++
|
||||
|
||||
return gate
|
||||
}
|
||||
|
||||
// ApproveGate approves a gate
|
||||
func (agm *ApprovalGateManager) ApproveGate(gateID, approvedBy, comments string) error {
|
||||
agm.mu.Lock()
|
||||
defer agm.mu.Unlock()
|
||||
|
||||
gate, exists := agm.gates[gateID]
|
||||
if !exists {
|
||||
return fmt.Errorf("gate not found: %s", gateID)
|
||||
}
|
||||
|
||||
if gate.Decision.Status == StatusApproved || gate.Decision.Status == StatusRejected {
|
||||
return fmt.Errorf("gate already has a decision: %s", gate.Decision.Status)
|
||||
}
|
||||
|
||||
if time.Now().After(gate.Decision.ExpiresAt) {
|
||||
gate.Decision.Status = StatusExpired
|
||||
agm.stats.ExpiredGates++
|
||||
agm.stats.PendingGates--
|
||||
return fmt.Errorf("gate has expired")
|
||||
}
|
||||
|
||||
gate.Decision.Status = StatusApproved
|
||||
gate.Decision.ApprovedBy = approvedBy
|
||||
gate.Decision.Comments = comments
|
||||
gate.Decision.Timestamp = time.Now()
|
||||
|
||||
gate.Approvals = append(gate.Approvals, approvedBy)
|
||||
|
||||
// Record in history
|
||||
record := &ApprovalRecord{
|
||||
GateID: gateID,
|
||||
Decision: StatusApproved,
|
||||
ApprovedBy: approvedBy,
|
||||
Timestamp: gate.Decision.Timestamp,
|
||||
Reason: comments,
|
||||
}
|
||||
|
||||
agm.history = append(agm.history, record)
|
||||
agm.stats.ApprovedGates++
|
||||
agm.stats.PendingGates--
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RejectGate rejects a gate
|
||||
func (agm *ApprovalGateManager) RejectGate(gateID, rejectedBy, reason string) error {
|
||||
agm.mu.Lock()
|
||||
defer agm.mu.Unlock()
|
||||
|
||||
gate, exists := agm.gates[gateID]
|
||||
if !exists {
|
||||
return fmt.Errorf("gate not found: %s", gateID)
|
||||
}
|
||||
|
||||
if gate.Decision.Status == StatusApproved || gate.Decision.Status == StatusRejected {
|
||||
return fmt.Errorf("gate already has a decision: %s", gate.Decision.Status)
|
||||
}
|
||||
|
||||
if time.Now().After(gate.Decision.ExpiresAt) {
|
||||
gate.Decision.Status = StatusExpired
|
||||
agm.stats.ExpiredGates++
|
||||
agm.stats.PendingGates--
|
||||
return fmt.Errorf("gate has expired")
|
||||
}
|
||||
|
||||
gate.Decision.Status = StatusRejected
|
||||
gate.Decision.RejectedBy = rejectedBy
|
||||
gate.Decision.Reason = reason
|
||||
gate.Decision.Timestamp = time.Now()
|
||||
|
||||
// Record in history
|
||||
record := &ApprovalRecord{
|
||||
GateID: gateID,
|
||||
Decision: StatusRejected,
|
||||
RejectedBy: rejectedBy,
|
||||
Timestamp: gate.Decision.Timestamp,
|
||||
Reason: reason,
|
||||
}
|
||||
|
||||
agm.history = append(agm.history, record)
|
||||
agm.stats.RejectedGates++
|
||||
agm.stats.PendingGates--
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetGate retrieves a gate
|
||||
func (agm *ApprovalGateManager) GetGate(gateID string) (*ApprovalGate, bool) {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
gate, exists := agm.gates[gateID]
|
||||
return gate, exists
|
||||
}
|
||||
|
||||
// GetDecision retrieves a decision
|
||||
func (agm *ApprovalGateManager) GetDecision(gateID string) (*ApprovalDecision, bool) {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
decision, exists := agm.decisions[gateID]
|
||||
return decision, exists
|
||||
}
|
||||
|
||||
// GetPendingGates returns all pending gates
|
||||
func (agm *ApprovalGateManager) GetPendingGates() []*ApprovalGate {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
pending := make([]*ApprovalGate, 0)
|
||||
for _, gate := range agm.gates {
|
||||
if gate.Decision.Status == StatusPending {
|
||||
// Check if expired
|
||||
if time.Now().After(gate.Decision.ExpiresAt) {
|
||||
gate.Decision.Status = StatusExpired
|
||||
} else {
|
||||
pending = append(pending, gate)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return pending
|
||||
}
|
||||
|
||||
// GetGatesByTask returns all gates for a task
|
||||
func (agm *ApprovalGateManager) GetGatesByTask(taskID string) []*ApprovalGate {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
gates := make([]*ApprovalGate, 0)
|
||||
for _, gate := range agm.gates {
|
||||
if gate.TaskID == taskID {
|
||||
gates = append(gates, gate)
|
||||
}
|
||||
}
|
||||
|
||||
return gates
|
||||
}
|
||||
|
||||
// GetGatesByWorkflow returns all gates for a workflow
|
||||
func (agm *ApprovalGateManager) GetGatesByWorkflow(workflowID string) []*ApprovalGate {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
gates := make([]*ApprovalGate, 0)
|
||||
for _, gate := range agm.gates {
|
||||
if gate.WorkflowID == workflowID {
|
||||
gates = append(gates, gate)
|
||||
}
|
||||
}
|
||||
|
||||
return gates
|
||||
}
|
||||
|
||||
// IsApproved checks if a gate is approved
|
||||
func (agm *ApprovalGateManager) IsApproved(gateID string) bool {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
gate, exists := agm.gates[gateID]
|
||||
if !exists {
|
||||
return false
|
||||
}
|
||||
|
||||
return gate.Decision.Status == StatusApproved
|
||||
}
|
||||
|
||||
// IsRejected checks if a gate is rejected
|
||||
func (agm *ApprovalGateManager) IsRejected(gateID string) bool {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
gate, exists := agm.gates[gateID]
|
||||
if !exists {
|
||||
return false
|
||||
}
|
||||
|
||||
return gate.Decision.Status == StatusRejected
|
||||
}
|
||||
|
||||
// IsPending checks if a gate is still pending
|
||||
func (agm *ApprovalGateManager) IsPending(gateID string) bool {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
gate, exists := agm.gates[gateID]
|
||||
if !exists {
|
||||
return false
|
||||
}
|
||||
|
||||
if time.Now().After(gate.Decision.ExpiresAt) {
|
||||
return false // Expired gates are not pending
|
||||
}
|
||||
|
||||
return gate.Decision.Status == StatusPending
|
||||
}
|
||||
|
||||
// GetStats returns statistics
|
||||
func (agm *ApprovalGateManager) GetStats() *ApprovalStats {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
stats := *agm.stats
|
||||
return &stats
|
||||
}
|
||||
|
||||
// GetHistory returns approval history
|
||||
func (agm *ApprovalGateManager) GetHistory() []*ApprovalRecord {
|
||||
agm.mu.RLock()
|
||||
defer agm.mu.RUnlock()
|
||||
|
||||
result := make([]*ApprovalRecord, len(agm.history))
|
||||
copy(result, agm.history)
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// Clear clears all gates
|
||||
func (agm *ApprovalGateManager) Clear() {
|
||||
agm.mu.Lock()
|
||||
defer agm.mu.Unlock()
|
||||
|
||||
agm.gates = make(map[string]*ApprovalGate)
|
||||
agm.decisions = make(map[string]*ApprovalDecision)
|
||||
agm.history = make([]*ApprovalRecord, 0)
|
||||
agm.stats = &ApprovalStats{}
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
package approval
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestNewApprovalGateManager(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
assert.NotNil(t, manager)
|
||||
assert.Equal(t, 0, manager.stats.TotalGates)
|
||||
}
|
||||
|
||||
func TestCreateGate(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
assert.NotNil(t, gate)
|
||||
assert.Equal(t, "T0.1", gate.TaskID)
|
||||
assert.Equal(t, StatusPending, gate.Decision.Status)
|
||||
}
|
||||
|
||||
func TestApproveGate(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
err := manager.ApproveGate(gate.ID, "reviewer-1", "Approved")
|
||||
assert.NoError(t, err)
|
||||
assert.True(t, manager.IsApproved(gate.ID))
|
||||
}
|
||||
|
||||
func TestRejectGate(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
err := manager.RejectGate(gate.ID, "reviewer-1", "Rejected")
|
||||
assert.NoError(t, err)
|
||||
assert.True(t, manager.IsRejected(gate.ID))
|
||||
}
|
||||
|
||||
func TestGetPendingGates(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate1 := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
gate2 := manager.CreateGate("T0.2", "workflow-1", "Review", 24*time.Hour)
|
||||
|
||||
manager.ApproveGate(gate1.ID, "reviewer-1", "")
|
||||
|
||||
pending := manager.GetPendingGates()
|
||||
assert.Equal(t, 1, len(pending))
|
||||
if len(pending) > 0 {
|
||||
assert.Equal(t, gate2.ID, pending[0].ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetGatesByTask(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
manager.CreateGate("T0.2", "workflow-1", "Review", 24*time.Hour)
|
||||
|
||||
gates := manager.GetGatesByTask("T0.1")
|
||||
assert.Greater(t, len(gates), 0)
|
||||
}
|
||||
|
||||
func TestGetGatesByWorkflow(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
manager.CreateGate("T0.2", "workflow-2", "Review", 24*time.Hour)
|
||||
|
||||
gates := manager.GetGatesByWorkflow("workflow-1")
|
||||
assert.Greater(t, len(gates), 0)
|
||||
}
|
||||
|
||||
func TestIsApproved(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
assert.False(t, manager.IsApproved(gate.ID))
|
||||
|
||||
manager.ApproveGate(gate.ID, "reviewer-1", "")
|
||||
assert.True(t, manager.IsApproved(gate.ID))
|
||||
}
|
||||
|
||||
func TestIsRejected(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
assert.False(t, manager.IsRejected(gate.ID))
|
||||
|
||||
manager.RejectGate(gate.ID, "reviewer-1", "")
|
||||
assert.True(t, manager.IsRejected(gate.ID))
|
||||
}
|
||||
|
||||
func TestIsPending(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
assert.True(t, manager.IsPending(gate.ID))
|
||||
|
||||
manager.ApproveGate(gate.ID, "reviewer-1", "")
|
||||
assert.False(t, manager.IsPending(gate.ID))
|
||||
}
|
||||
|
||||
func TestGetStats(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate1 := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
_ = manager.CreateGate("T0.2", "workflow-1", "Review", 24*time.Hour)
|
||||
|
||||
manager.ApproveGate(gate1.ID, "reviewer-1", "")
|
||||
|
||||
stats := manager.GetStats()
|
||||
assert.Equal(t, 2, stats.TotalGates)
|
||||
assert.Equal(t, 1, stats.ApprovedGates)
|
||||
assert.Equal(t, 1, stats.PendingGates)
|
||||
}
|
||||
|
||||
func TestGetHistory(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate1 := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
gate2 := manager.CreateGate("T0.2", "workflow-1", "Review", 24*time.Hour)
|
||||
|
||||
manager.ApproveGate(gate1.ID, "reviewer-1", "")
|
||||
manager.RejectGate(gate2.ID, "reviewer-2", "Needs work")
|
||||
|
||||
history := manager.GetHistory()
|
||||
assert.Greater(t, len(history), 0)
|
||||
}
|
||||
|
||||
func TestClear(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
assert.Equal(t, 1, manager.stats.TotalGates)
|
||||
|
||||
manager.Clear()
|
||||
assert.Equal(t, 0, manager.stats.TotalGates)
|
||||
}
|
||||
|
||||
func TestExpiredGate(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 1*time.Millisecond)
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
err := manager.ApproveGate(gate.ID, "reviewer-1", "")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestMultipleApprovals(t *testing.T) {
|
||||
manager := NewApprovalGateManager()
|
||||
gate := manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
gate.RequiredApprovals = 2
|
||||
|
||||
err := manager.ApproveGate(gate.ID, "reviewer-1", "")
|
||||
assert.NoError(t, err)
|
||||
|
||||
retrieved, _ := manager.GetGate(gate.ID)
|
||||
assert.Equal(t, 1, len(retrieved.Approvals))
|
||||
}
|
||||
|
||||
func BenchmarkCreateGate(b *testing.B) {
|
||||
manager := NewApprovalGateManager()
|
||||
for i := 0; i < b.N; i++ {
|
||||
manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkApproveGate(b *testing.B) {
|
||||
manager := NewApprovalGateManager()
|
||||
gates := make([]*ApprovalGate, b.N)
|
||||
for i := 0; i < b.N; i++ {
|
||||
gates[i] = manager.CreateGate("T0.1", "workflow-1", "Review", 24*time.Hour)
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
manager.ApproveGate(gates[i].ID, "reviewer-1", "")
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -7,7 +7,7 @@
|
||||
| T3.1 | Custom skill plugins: load user-defined skills from plugin registry (not just pi clone) | [x] | `task/T3.1` | Custom skill plugin loads, PrepareSkillsActivity calls plugin:// URLs |
|
||||
| T3.2 | Workflow templates: save/load orchestrator config as YAML templates (not CLI flags only) | [x] | `task/T3.2` | Load template `templates/golang-project.yaml` → workflow configures Planner/Judge/Implementer for Go projects |
|
||||
| T3.3 | Task dependency graph: specify task order (T0.2 must complete before T0.3 can start) | [x] | `task/T3.3` | Board supports `depends_on: [T0.1]` field, orchestrator respects ordering |
|
||||
| T3.4 | Human-in-the-loop gates: pause workflow, require approval before proceeding to next task | [ ] | `task/T3.4` | Workflow waits for `approve-task` signal, Judge verdict is final (can't auto-retry after user approval) |
|
||||
| T3.4 | Human-in-the-loop gates: pause workflow, require approval before proceeding to next task | [x] | `task/T3.4` | Workflow waits for `approve-task` signal, Judge verdict is final (can't auto-retry after user approval) |
|
||||
| T3.5 | Custom Judge implementations: swap default Judge for domain-specific validator (e.g., security auditor) | [ ] | `task/T3.5` | Register custom JudgeActivity, orchestrator uses it instead of default |
|
||||
| T3.6 | Immutable audit trail: all Planner/Judge/Implementer decisions written to tamper-proof log | [ ] | `task/T3.6` | Audit log signed with per-workflow key, verification prevents tampering |
|
||||
| T3.7 | Workflow composition: nest OrchestratorWorkflows (one orchestrator dispatches child orchestrators) | [ ] | `task/T3.7` | Multi-level task hierarchy: T0 milestone → T0.a/T0.b sub-milestones, each with own orchestrator |
|
||||
|
||||
Reference in New Issue
Block a user