merge: universal event architecture

Generalize mecha's event system from git-specific to provider-neutral,
enabling any webhook source, active trigger, or scheduled event to flow
through the same pipeline.

Architecture:
- Event struct: Actor, Subject, Attrs (replaces RepoOwner/Name/Ref/Number/Sender)
- 5 provider interfaces: Source, Trigger, Hydrator, Verifier, Responder
- Authenticated marker for webhook auth bypass control
- Responder registry for provider-agnostic write-back
- DB migration v3 (transactional table swap)
- Crash recovery for stuck received events

Quality:
- 23 Codex audit findings fixed (1 critical, 8 high, 9 medium, 5 low)
- 5 grill paranoid-mode agents, all findings addressed
- 3 bugs found and fixed from cross-round analysis
- 310 tests (was 166), serve coverage 77.2% (was 55%)
- Disposable container lifecycle tested with real Docker
- Documentation audit: all 4 dimensions at 100%

13 commits, 34 files changed

Author: xiaolai

.claude/rules/domain-model.md

@@ -10,12 +10,35 @@ Mecha has exactly four nouns: Event, Worker, Task, Policy.
 ## Pipeline
 
 ```
-Event.arrive → Event.match → Task.create → Task.dispatch → Policy.filter → Task.complete
+Event.arrive → Event.match → Task.create → Task.dispatch → Policy.filter → Respond
 ```
 
+## Event Model (Universal)
+
+Events are provider-neutral. All provider-specific data lives in `Attrs`.
+
+| Field | Purpose | Example |
+|---|---|---|
+| `Source` | Provider name | `github`, `slack`, `telegram`, `cron` |
+| `Type` | Event type | `pull_request.opened`, `message`, `tick` |
+| `Actor` | Who triggered | username, phone, bot name |
+| `Subject` | What it's about | `owner/repo`, `#channel`, `schedule-daily` |
+| `Attrs` | Provider-specific fields | `repo_owner`, `number`, `diff`, `text` |
+| `DedupKey` | Semantic dedup | Content hash for polls/cron |
+
+## Provider Interfaces
+
+| Interface | Direction | Purpose |
+|---|---|---|
+| `Source` | Inbound (passive) | Parse webhooks into Events |
+| `Trigger` | Inbound (active) | Generate events (cron, polling) |
+| `Hydrator` | Enrichment | Fetch additional data via API |
+| `Verifier` | Handshake | Challenge-response verification |
+| `Responder` | Outbound | Write results back to platform |
+
 ## Nouns and Verbs
 
-- **Event**: arrive, match
+- **Event**: arrive, match, hydrate, skip, fail
 - **Worker**: add, remove, start, stop, ls
 - **Task**: create, dispatch, complete, fail
 - **Policy**: filter
@@ -28,3 +51,4 @@ Event.arrive → Event.match → Task.create → Task.dispatch → Policy.filter
 - One noun, one lifecycle.
 - No hidden nouns. Unowned logic means a missing noun.
 - Verbs are idempotent where possible.
+- Responder is keyed by target, not source (ingress != egress).

.claude/rules/result-contract.md

@@ -5,7 +5,7 @@ globs: "**/*.go"
 
 # Result Contract
 
-> Design spec. Implemented in TypeScript (`docker/runtime/types.ts`). No Go struct yet — Phase 3.
+> Implemented in Go (`internal/policy/result.go`) and TypeScript (`docker/runtime/types.ts`).
 
 Every worker returns the same structure. Every field optional.
 
@@ -25,9 +25,19 @@ Every worker returns the same structure. Every field optional.
 }
 ```
 
+## Write-Back
+
+Write-back is routed through the Responder registry, keyed by target platform:
+
+- `source.Responder` interface: `Name() string`, `Respond(ctx, ev, result) error`
+- GitHub writeback.Client implements Responder (registered automatically)
+- Dispatch tries Responder registry first, falls back to legacy writeback
+- Responder is looked up by `ev.Source` (e.g., GitHub events use the GitHub responder)
+
 ## Rules
 
 - Result and side effects are one thing. No separate "action" phase.
 - Policy filters the result before write-back. Denied fields are dropped.
 - A completed task means its result has already been written back.
 - The worker decides what to return. Policy decides what gets through.
+- Responder is looked up by `ev.Source`. Target override is planned but not yet implemented.

internal/cli/serve.go

@@ -78,10 +78,11 @@ func serveCmd() *cobra.Command {
 				logger.Info("gitlab source registered")
 			}
 
-			// Write-back client
+			// Write-back client + responder registration
 			var wb *writeback.Client
 			if ghToken != "" {
 				wb = writeback.NewClient(ghToken, logger)
+				sources.RegisterResponder(wb)
 			}
 
 			srv := serve.New(serve.Config{

internal/event/helpers.go

@@ -13,11 +13,11 @@ type rowScanner interface{ Scan(dest ...any) error }
 
 func scanEvent(s rowScanner) (*Event, error) {
 	var ev Event
-	var state, payloadStr, raw, wn, tid string
+	var state, attrsStr, raw, wn, tid string
 	var createdAt, updatedAt int64
-	err := s.Scan(&ev.ID, &ev.DeliveryID, &ev.Source, &ev.Type,
-		&ev.RepoOwner, &ev.RepoName, &ev.Ref, &ev.Number, &ev.Sender,
-		&payloadStr, &raw, &state, &wn, &tid, &createdAt, &updatedAt)
+	err := s.Scan(&ev.ID, &ev.DeliveryID, &ev.DedupKey,
+		&ev.Source, &ev.Type, &ev.Actor, &ev.Subject,
+		&attrsStr, &raw, &state, &wn, &tid, &createdAt, &updatedAt)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, fmt.Errorf("event not found")
@@ -30,11 +30,14 @@ func scanEvent(s rowScanner) (*Event, error) {
 	ev.Raw = json.RawMessage(raw)
 	ev.CreatedAt = time.Unix(createdAt, 0)
 	ev.UpdatedAt = time.Unix(updatedAt, 0)
-	if payloadStr != "" {
-		if err := json.Unmarshal([]byte(payloadStr), &ev.Payload); err != nil {
-			return nil, fmt.Errorf("unmarshal event payload: %w", err)
+	if attrsStr != "" && attrsStr != "null" {
+		if err := json.Unmarshal([]byte(attrsStr), &ev.Attrs); err != nil {
+			return nil, fmt.Errorf("unmarshal event attrs: %w", err)
 		}
 	}
+	if ev.Attrs == nil {
+		ev.Attrs = make(Attrs)
+	}
 	return &ev, nil
 }
 

internal/event/store.go

@@ -18,6 +18,9 @@ func NewStore(db *sql.DB) *Store {
 	return &Store{db: db}
 }
 
+const eventColumns = `id, delivery_id, dedup_key, source, type, actor, subject,
+	attrs, raw, state, worker_name, task_id, created_at, updated_at`
+
 // Create persists a new event in received state.
 func (s *Store) Create(ctx context.Context, ev *Event) error {
 	if ev.ID == "" {
@@ -32,17 +35,17 @@ func (s *Store) Create(ctx context.Context, ev *Event) error {
 	ev.CreatedAt = now
 	ev.UpdatedAt = now
 
-	payload, err := json.Marshal(ev.Payload)
+	attrs, err := json.Marshal(ev.Attrs)
 	if err != nil {
-		return fmt.Errorf("marshal event payload: %w", err)
+		return fmt.Errorf("marshal event attrs: %w", err)
 	}
 
 	_, err = s.db.ExecContext(ctx,
-		`INSERT INTO events (id, delivery_id, source, type, repo_owner, repo_name,
-		 ref, number, sender, payload, raw, state, created_at, updated_at)
-		 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
-		ev.ID, ev.DeliveryID, ev.Source, ev.Type, ev.RepoOwner, ev.RepoName,
-		ev.Ref, ev.Number, ev.Sender, string(payload), string(ev.Raw),
+		`INSERT INTO events (id, delivery_id, dedup_key, source, type, actor, subject,
+		 attrs, raw, state, created_at, updated_at)
+		 VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`,
+		ev.ID, ev.DeliveryID, ev.DedupKey, ev.Source, ev.Type, ev.Actor, ev.Subject,
+		string(attrs), string(ev.Raw),
 		string(ev.State), now.Unix(), now.Unix(),
 	)
 	if err != nil {
@@ -69,17 +72,13 @@ func (s *Store) DeliveryExists(ctx context.Context, deliveryID string) (bool, er
 // Get retrieves an event by ID.
 func (s *Store) Get(ctx context.Context, id string) (*Event, error) {
 	row := s.db.QueryRowContext(ctx,
-		`SELECT id, delivery_id, source, type, repo_owner, repo_name,
-		 ref, number, sender, payload, raw, state, worker_name, task_id,
-		 created_at, updated_at FROM events WHERE id = ?`, id)
+		`SELECT `+eventColumns+` FROM events WHERE id = ?`, id)
 	return scanEvent(row)
 }
 
 // List returns events, optionally filtered by state.
 func (s *Store) List(ctx context.Context, state string) ([]Event, error) {
-	query := `SELECT id, delivery_id, source, type, repo_owner, repo_name,
-	          ref, number, sender, payload, raw, state, worker_name, task_id,
-	          created_at, updated_at FROM events`
+	query := `SELECT ` + eventColumns + ` FROM events`
 	var args []any
 	if state != "" {
 		query += " WHERE state = ?"
@@ -102,6 +101,22 @@ func (s *Store) List(ctx context.Context, state string) ([]Event, error) {
 	return events, rows.Err()
 }
 
+// UpdateAttrs persists hydrated attrs back to the database.
+func (s *Store) UpdateAttrs(ctx context.Context, id string, attrs Attrs) error {
+	data, err := json.Marshal(attrs)
+	if err != nil {
+		return fmt.Errorf("marshal attrs: %w", err)
+	}
+	now := time.Now().Unix()
+	res, err := s.db.ExecContext(ctx,
+		"UPDATE events SET attrs = ?, updated_at = ? WHERE id = ?",
+		string(data), now, id)
+	if err != nil {
+		return fmt.Errorf("update attrs: %w", err)
+	}
+	return checkRows(res, id, "update attrs")
+}
+
 // SetMatched transitions received → matched with worker name.
 func (s *Store) SetMatched(ctx context.Context, id, workerName string) error {
 	return s.transitionWithExtra(ctx, id, StateReceived, StateMatched, workerName,
@@ -138,6 +153,27 @@ func (s *Store) SetSkipped(ctx context.Context, id string) error {
 		"UPDATE events SET state = ?, updated_at = ? WHERE id = ? AND state = ?")
 }
 
+// Received returns IDs of events stuck in received state (for crash recovery).
+func (s *Store) Received(ctx context.Context) ([]string, error) {
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT id FROM events WHERE state = ? ORDER BY created_at`,
+		string(StateReceived),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("query received: %w", err)
+	}
+	defer rows.Close()
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, rows.Err()
+}
+
 func (s *Store) transition(ctx context.Context, id string, from, to State, query string) error {
 	now := time.Now().Unix()
 	res, err := s.db.ExecContext(ctx, query, string(to), now, id, string(from))
@@ -155,4 +191,3 @@ func (s *Store) transitionWithExtra(ctx context.Context, id string, from, to Sta
 	}
 	return checkRows(res, id, string(to))
 }
-

internal/event/store_test.go

@@ -22,14 +22,12 @@ func testStore(t *testing.T) *Store {
 func TestCreateAndGet(t *testing.T) {
 	s := testStore(t)
 	ev := &Event{
-		Source:    "github",
-		Type:      "pull_request.opened",
-		RepoOwner: "owner",
-		RepoName:  "repo",
-		Number:    42,
-		Sender:    "user",
-		Payload:   Payload{"title": "Fix bug"},
-		Raw:       json.RawMessage(`{"action":"opened"}`),
+		Source:  "github",
+		Type:    "pull_request.opened",
+		Actor:   "user",
+		Subject: "owner/repo",
+		Attrs:   Attrs{"title": "Fix bug", "number": 42, "repo_owner": "owner", "repo_name": "repo"},
+		Raw:     json.RawMessage(`{"action":"opened"}`),
 	}
 	if err := s.Create(context.Background(), ev); err != nil {
 		t.Fatal(err)
@@ -48,8 +46,14 @@ func TestCreateAndGet(t *testing.T) {
 	if got.State != StateReceived {
 		t.Errorf("state = %q", got.State)
 	}
-	if got.Payload["title"] != "Fix bug" {
-		t.Errorf("payload = %v", got.Payload)
+	if got.Attrs["title"] != "Fix bug" {
+		t.Errorf("attrs = %v", got.Attrs)
+	}
+	if got.Actor != "user" {
+		t.Errorf("actor = %q", got.Actor)
+	}
+	if got.Subject != "owner/repo" {
+		t.Errorf("subject = %q", got.Subject)
 	}
 }
 
@@ -137,3 +141,37 @@ func TestGetNotFound(t *testing.T) {
 		t.Error("expected error")
 	}
 }
+
+func TestUpdateAttrs(t *testing.T) {
+	s := testStore(t)
+	ctx := context.Background()
+	ev := &Event{Source: "github", Type: "push", Attrs: Attrs{"a": "1"}}
+	s.Create(ctx, ev)
+
+	err := s.UpdateAttrs(ctx, ev.ID, Attrs{"a": "1", "diff": "some diff"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, _ := s.Get(ctx, ev.ID)
+	if got.Attrs["diff"] != "some diff" {
+		t.Errorf("attrs = %v", got.Attrs)
+	}
+}
+
+func TestReceived(t *testing.T) {
+	s := testStore(t)
+	ctx := context.Background()
+	ev1 := &Event{Source: "github", Type: "push"}
+	ev2 := &Event{Source: "slack", Type: "message"}
+	s.Create(ctx, ev1)
+	s.Create(ctx, ev2)
+	s.SetMatched(ctx, ev2.ID, "w")
+
+	ids, err := s.Received(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(ids) != 1 || ids[0] != ev1.ID {
+		t.Errorf("received = %v, want [%s]", ids, ev1.ID)
+	}
+}

internal/event/types.go

@@ -9,29 +9,35 @@ import (
 type State string
 
 const (
-	StateReceived  State = "received"
-	StateMatched   State = "matched"
+	// StateReceived means the event arrived and was persisted, not yet matched.
+	StateReceived State = "received"
+	// StateMatched means the event matched a worker by event rules.
+	StateMatched State = "matched"
+	// StateDispatched means a task was created and sent to a worker.
 	StateDispatched State = "dispatched"
+	// StateCompleted means the worker returned a result and write-back succeeded.
 	StateCompleted State = "completed"
-	StateFailed    State = "failed"
-	StateSkipped   State = "skipped"
+	// StateFailed means processing errored or timed out.
+	StateFailed State = "failed"
+	// StateSkipped means no matching worker was found for this event.
+	StateSkipped State = "skipped"
 )
 
-// Payload is a map of enriched fields available to prompt templates.
-type Payload map[string]any
+// Attrs holds provider-specific fields available to prompt templates.
+type Attrs map[string]any
 
 // Event is something that happened from an external source.
+// All provider-specific data lives in Attrs — the struct itself
+// contains only universal, provider-neutral fields.
 type Event struct {
 	ID         string          `json:"id"`
 	DeliveryID string          `json:"delivery_id"`
+	DedupKey   string          `json:"dedup_key,omitempty"` // reserved for semantic dedup (cron/polling); not yet enforced
 	Source     string          `json:"source"`
 	Type       string          `json:"type"`
-	RepoOwner  string          `json:"repo_owner"`
-	RepoName   string          `json:"repo_name"`
-	Ref        string          `json:"ref"`
-	Number     int             `json:"number"`
-	Sender     string          `json:"sender"`
-	Payload    Payload         `json:"payload"`
+	Actor      string          `json:"actor"`
+	Subject    string          `json:"subject"`
+	Attrs      Attrs           `json:"attrs"`
 	Raw        json.RawMessage `json:"raw,omitempty"`
 	State      State           `json:"state"`
 	WorkerName string          `json:"worker_name,omitempty"`