feat(mail): implement beads-native gt mail claim command
Implement claiming for queue messages using beads-native approach: - Add claim_pattern field to QueueFields for eligibility checking - Add MatchClaimPattern function for pattern matching (wildcards supported) - Add FindEligibleQueues to find all queues an agent can claim from - Rewrite runMailClaim to use beads-native queue lookup - Support optional queue argument (claim from any eligible if not specified) - Use claimed-by/claimed-at labels instead of changing assignee - Update runMailRelease to work with new claiming approach - Add comprehensive tests for pattern matching and validation Queue messages are now claimed via labels: - claimed-by: <agent-identity> - claimed-at: <RFC3339 timestamp> Messages with queue:<name> label but no claimed-by are unclaimed. Closes gt-xfqh1e.11 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gastown/crew/jack
committed by
Steve Yegge
Steve Yegge
parent
012d50b2b2
commit
2ffc8e8712
@@ -14,6 +14,7 @@ import (
|
||||
// These are stored as "key: value" lines in the description.
|
||||
type QueueFields struct {
|
||||
Name string // Queue name (human-readable identifier)
|
||||
ClaimPattern string // Pattern for who can claim from queue (e.g., "gastown/polecats/*")
|
||||
Status string // active, paused, closed
|
||||
MaxConcurrency int // Maximum number of concurrent workers (0 = unlimited)
|
||||
ProcessingOrder string // fifo, priority (default: fifo)
|
||||
@@ -52,6 +53,12 @@ func FormatQueueDescription(title string, fields *QueueFields) string {
|
||||
lines = append(lines, "name: null")
|
||||
}
|
||||
|
||||
if fields.ClaimPattern != "" {
|
||||
lines = append(lines, fmt.Sprintf("claim_pattern: %s", fields.ClaimPattern))
|
||||
} else {
|
||||
lines = append(lines, "claim_pattern: *") // Default: anyone can claim
|
||||
}
|
||||
|
||||
if fields.Status != "" {
|
||||
lines = append(lines, fmt.Sprintf("status: %s", fields.Status))
|
||||
} else {
|
||||
@@ -79,6 +86,7 @@ func ParseQueueFields(description string) *QueueFields {
|
||||
fields := &QueueFields{
|
||||
Status: QueueStatusActive,
|
||||
ProcessingOrder: QueueOrderFIFO,
|
||||
ClaimPattern: "*", // Default: anyone can claim
|
||||
}
|
||||
|
||||
for _, line := range strings.Split(description, "\n") {
|
||||
@@ -101,6 +109,10 @@ func ParseQueueFields(description string) *QueueFields {
|
||||
switch strings.ToLower(key) {
|
||||
case "name":
|
||||
fields.Name = value
|
||||
case "claim_pattern":
|
||||
if value != "" {
|
||||
fields.ClaimPattern = value
|
||||
}
|
||||
case "status":
|
||||
fields.Status = value
|
||||
case "max_concurrency":
|
||||
@@ -266,3 +278,72 @@ func (b *Beads) DeleteQueueBead(id string) error {
|
||||
_, err := b.run("delete", id, "--hard", "--force")
|
||||
return err
|
||||
}
|
||||
|
||||
// MatchClaimPattern checks if an identity matches a claim pattern.
|
||||
// Patterns support:
|
||||
// - "*" matches anyone
|
||||
// - "gastown/polecats/*" matches any polecat in gastown rig
|
||||
// - "*/witness" matches any witness role across rigs
|
||||
// - Exact match for specific identities
|
||||
func MatchClaimPattern(pattern, identity string) bool {
|
||||
// Wildcard matches anyone
|
||||
if pattern == "*" {
|
||||
return true
|
||||
}
|
||||
|
||||
// Exact match
|
||||
if pattern == identity {
|
||||
return true
|
||||
}
|
||||
|
||||
// Wildcard pattern matching
|
||||
if strings.Contains(pattern, "*") {
|
||||
// Convert to simple glob matching
|
||||
// "gastown/polecats/*" should match "gastown/polecats/capable"
|
||||
// "*/witness" should match "gastown/witness"
|
||||
parts := strings.Split(pattern, "*")
|
||||
if len(parts) == 2 {
|
||||
prefix := parts[0]
|
||||
suffix := parts[1]
|
||||
if strings.HasPrefix(identity, prefix) && strings.HasSuffix(identity, suffix) {
|
||||
// Check that the middle part doesn't contain path separators
|
||||
// unless the pattern allows it (e.g., "*/" at start)
|
||||
middle := identity[len(prefix) : len(identity)-len(suffix)]
|
||||
// Only allow single segment match (no extra slashes)
|
||||
if !strings.Contains(middle, "/") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// FindEligibleQueues returns all queue beads that the given identity can claim from.
|
||||
func (b *Beads) FindEligibleQueues(identity string) ([]*Issue, []*QueueFields, error) {
|
||||
queues, err := b.ListQueueBeads()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
var eligibleIssues []*Issue
|
||||
var eligibleFields []*QueueFields
|
||||
|
||||
for _, issue := range queues {
|
||||
fields := ParseQueueFields(issue.Description)
|
||||
|
||||
// Skip inactive queues
|
||||
if fields.Status != QueueStatusActive {
|
||||
continue
|
||||
}
|
||||
|
||||
// Check if identity matches claim pattern
|
||||
if MatchClaimPattern(fields.ClaimPattern, identity) {
|
||||
eligibleIssues = append(eligibleIssues, issue)
|
||||
eligibleFields = append(eligibleFields, fields)
|
||||
}
|
||||
}
|
||||
|
||||
return eligibleIssues, eligibleFields, nil
|
||||
}
|
||||
|
||||
301
internal/beads/beads_queue_test.go
Normal file
301
internal/beads/beads_queue_test.go
Normal file
@@ -0,0 +1,301 @@
|
||||
package beads
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMatchClaimPattern(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
pattern string
|
||||
identity string
|
||||
want bool
|
||||
}{
|
||||
// Wildcard matches anyone
|
||||
{
|
||||
name: "wildcard matches anyone",
|
||||
pattern: "*",
|
||||
identity: "gastown/crew/max",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "wildcard matches town-level agent",
|
||||
pattern: "*",
|
||||
identity: "mayor/",
|
||||
want: true,
|
||||
},
|
||||
|
||||
// Exact match
|
||||
{
|
||||
name: "exact match",
|
||||
pattern: "gastown/crew/max",
|
||||
identity: "gastown/crew/max",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "exact match fails on different identity",
|
||||
pattern: "gastown/crew/max",
|
||||
identity: "gastown/crew/nux",
|
||||
want: false,
|
||||
},
|
||||
|
||||
// Suffix wildcard
|
||||
{
|
||||
name: "suffix wildcard matches",
|
||||
pattern: "gastown/polecats/*",
|
||||
identity: "gastown/polecats/capable",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "suffix wildcard matches different name",
|
||||
pattern: "gastown/polecats/*",
|
||||
identity: "gastown/polecats/nux",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "suffix wildcard doesn't match nested path",
|
||||
pattern: "gastown/polecats/*",
|
||||
identity: "gastown/polecats/sub/capable",
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "suffix wildcard doesn't match different rig",
|
||||
pattern: "gastown/polecats/*",
|
||||
identity: "bartertown/polecats/capable",
|
||||
want: false,
|
||||
},
|
||||
|
||||
// Prefix wildcard
|
||||
{
|
||||
name: "prefix wildcard matches",
|
||||
pattern: "*/witness",
|
||||
identity: "gastown/witness",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "prefix wildcard matches different rig",
|
||||
pattern: "*/witness",
|
||||
identity: "bartertown/witness",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "prefix wildcard doesn't match different role",
|
||||
pattern: "*/witness",
|
||||
identity: "gastown/refinery",
|
||||
want: false,
|
||||
},
|
||||
|
||||
// Crew patterns
|
||||
{
|
||||
name: "crew wildcard",
|
||||
pattern: "gastown/crew/*",
|
||||
identity: "gastown/crew/max",
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "crew wildcard matches any crew member",
|
||||
pattern: "gastown/crew/*",
|
||||
identity: "gastown/crew/jack",
|
||||
want: true,
|
||||
},
|
||||
|
||||
// Edge cases
|
||||
{
|
||||
name: "empty identity doesn't match",
|
||||
pattern: "*",
|
||||
identity: "",
|
||||
want: true, // * matches anything
|
||||
},
|
||||
{
|
||||
name: "empty pattern doesn't match",
|
||||
pattern: "",
|
||||
identity: "gastown/crew/max",
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := MatchClaimPattern(tt.pattern, tt.identity)
|
||||
if got != tt.want {
|
||||
t.Errorf("MatchClaimPattern(%q, %q) = %v, want %v",
|
||||
tt.pattern, tt.identity, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFormatQueueDescription(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
title string
|
||||
fields *QueueFields
|
||||
want []string // Lines that should be present
|
||||
}{
|
||||
{
|
||||
name: "basic queue",
|
||||
title: "Queue: work-requests",
|
||||
fields: &QueueFields{
|
||||
Name: "work-requests",
|
||||
ClaimPattern: "gastown/crew/*",
|
||||
Status: QueueStatusActive,
|
||||
},
|
||||
want: []string{
|
||||
"Queue: work-requests",
|
||||
"name: work-requests",
|
||||
"claim_pattern: gastown/crew/*",
|
||||
"status: active",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "queue with default claim pattern",
|
||||
title: "Queue: public",
|
||||
fields: &QueueFields{
|
||||
Name: "public",
|
||||
Status: QueueStatusActive,
|
||||
},
|
||||
want: []string{
|
||||
"name: public",
|
||||
"claim_pattern: *", // Default
|
||||
"status: active",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "queue with counts",
|
||||
title: "Queue: processing",
|
||||
fields: &QueueFields{
|
||||
Name: "processing",
|
||||
ClaimPattern: "*/refinery",
|
||||
Status: QueueStatusActive,
|
||||
AvailableCount: 5,
|
||||
ProcessingCount: 2,
|
||||
CompletedCount: 10,
|
||||
FailedCount: 1,
|
||||
},
|
||||
want: []string{
|
||||
"name: processing",
|
||||
"claim_pattern: */refinery",
|
||||
"available_count: 5",
|
||||
"processing_count: 2",
|
||||
"completed_count: 10",
|
||||
"failed_count: 1",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "nil fields",
|
||||
title: "Just Title",
|
||||
fields: nil,
|
||||
want: []string{"Just Title"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := FormatQueueDescription(tt.title, tt.fields)
|
||||
for _, line := range tt.want {
|
||||
if !strings.Contains(got, line) {
|
||||
t.Errorf("FormatQueueDescription() missing line %q in:\n%s", line, got)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseQueueFields(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
description string
|
||||
wantName string
|
||||
wantPattern string
|
||||
wantStatus string
|
||||
}{
|
||||
{
|
||||
name: "basic queue",
|
||||
description: `Queue: work-requests
|
||||
|
||||
name: work-requests
|
||||
claim_pattern: gastown/crew/*
|
||||
status: active`,
|
||||
wantName: "work-requests",
|
||||
wantPattern: "gastown/crew/*",
|
||||
wantStatus: QueueStatusActive,
|
||||
},
|
||||
{
|
||||
name: "queue with defaults",
|
||||
description: `Queue: minimal
|
||||
|
||||
name: minimal`,
|
||||
wantName: "minimal",
|
||||
wantPattern: "*", // Default
|
||||
wantStatus: QueueStatusActive,
|
||||
},
|
||||
{
|
||||
name: "empty description",
|
||||
description: "",
|
||||
wantName: "",
|
||||
wantPattern: "*", // Default
|
||||
wantStatus: QueueStatusActive,
|
||||
},
|
||||
{
|
||||
name: "queue with counts",
|
||||
description: `Queue: processing
|
||||
|
||||
name: processing
|
||||
claim_pattern: */refinery
|
||||
status: paused
|
||||
available_count: 5
|
||||
processing_count: 2`,
|
||||
wantName: "processing",
|
||||
wantPattern: "*/refinery",
|
||||
wantStatus: QueueStatusPaused,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := ParseQueueFields(tt.description)
|
||||
if got.Name != tt.wantName {
|
||||
t.Errorf("Name = %q, want %q", got.Name, tt.wantName)
|
||||
}
|
||||
if got.ClaimPattern != tt.wantPattern {
|
||||
t.Errorf("ClaimPattern = %q, want %q", got.ClaimPattern, tt.wantPattern)
|
||||
}
|
||||
if got.Status != tt.wantStatus {
|
||||
t.Errorf("Status = %q, want %q", got.Status, tt.wantStatus)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueueBeadID(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
queueName string
|
||||
isTownLevel bool
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "town-level queue",
|
||||
queueName: "dispatch",
|
||||
isTownLevel: true,
|
||||
want: "hq-q-dispatch",
|
||||
},
|
||||
{
|
||||
name: "rig-level queue",
|
||||
queueName: "merge",
|
||||
isTownLevel: false,
|
||||
want: "gt-q-merge",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := QueueBeadID(tt.queueName, tt.isTownLevel)
|
||||
if got != tt.want {
|
||||
t.Errorf("QueueBeadID(%q, %v) = %q, want %q",
|
||||
tt.queueName, tt.isTownLevel, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user