feat(mcp): record which agent changed tasks and pipeline stages

CLAUDE.md

@@ -12,6 +12,7 @@ This file is intentionally slim. Read the doc(s) relevant to what you're touchin
 - [docs/commands.md](docs/commands.md) - full CLI command reference, config.yaml schema
 - [docs/broker-replies.md](docs/broker-replies.md) - reading data-broker email replies and turning each response type into the right brokers.yaml edit
 - [docs/multi-profile.md](docs/multi-profile.md) - the `--profile`/`profiles:` feature: config model, history scoping, shared-inbox attribution, web UI switcher
+- [docs/mcp.md](docs/mcp.md) - the experimental `eraser mcp` agent server: batch approval, shared-platform host scope, and what it may and may not do
 - [docs/code-patterns.md](docs/code-patterns.md) - conventions to follow, plus known quirks/gotchas worth reading before touching related code
 - [docs/auditing.md](docs/auditing.md) - how to re-run the dead-code/security sweep, and what's already been removed
 - [docs/stability.md](docs/stability.md) - what 1.x promises not to break (config keys, CLI flags, history.db, brokers.yaml, export JSON) - check before renaming or removing any of them

docs/mcp.md

@@ -130,7 +130,9 @@ page, mailbox, and previous attempts before proposing a retry. An observation
 does not release that safeguard. Native invalid forms can be corrected without
 submitting them; server-side rejection or ambiguous receipts stay open for
 review. The current receipt check is conservative and recognizes English
-acceptance text; other languages may need manual review.
+acceptance text; other languages may need manual review. The agent quotes the
+receipt text from an untrusted page, so a page can phrase a fake receipt that
+the check accepts. The evidence is recorded for you to read, not as proof.
 
 For CAPTCHA, login, identity documents, or missing profile information, the
 agent stops. Complete the step in the visible browser, then ask it to

internal/evidence/evidence.go

@@ -48,6 +48,7 @@ type BrokerEvidence struct {
 	BrokerEmail    string             `json:"broker_email,omitempty"`
 	BrokerWebsite  string             `json:"broker_website,omitempty"`
 	PipelineStatus string             `json:"pipeline_status,omitempty"`
+	PipelineSetBy  string             `json:"pipeline_set_by,omitempty"`
 	PastDeadline   bool               `json:"past_deadline"`
 	DeadlineDate   *time.Time         `json:"deadline_date,omitempty"`
 	Requests       []RequestEvidence  `json:"requests"`
@@ -183,6 +184,7 @@ func Build(
 		}
 
 		var latestPipeline history.PipelineStatus
+		var latestActor string
 		var earliestSent time.Time
 		var anySent bool
 		for _, req := range reqs {
@@ -204,7 +206,7 @@ func Build(
 			}
 			be.Requests = append(be.Requests, re)
 			if req.PipelineStatus != "" {
-				latestPipeline = req.PipelineStatus
+				latestPipeline, latestActor = req.PipelineStatus, req.PipelineActor
 			}
 			if req.Status == history.StatusSent && !req.SentAt.IsZero() {
 				if !anySent || req.SentAt.Before(earliestSent) {
@@ -214,6 +216,7 @@ func Build(
 			}
 		}
 		be.PipelineStatus = string(latestPipeline)
+		be.PipelineSetBy = latestActor
 
 		resp := responsesByBroker[brokerID]
 		sort.SliceStable(resp, func(i, j int) bool { return resp[i].ReceivedAt.Before(resp[j].ReceivedAt) })

internal/history/actor_test.go

@@ -0,0 +1,71 @@
+package history
+
+import (
+	"testing"
+	"time"
+)
+
+func TestTaskActorAndReviewReasonRoundTrip(t *testing.T) {
+	s := newTestStore(t)
+	task := &PendingTask{ProfileID: "default", BrokerID: "acme", BrokerName: "Acme", TaskType: TaskManualForm, Actor: "mcp:codex", ReviewReason: "needs_user"}
+	if err := s.AddPendingTask(task); err != nil {
+		t.Fatal(err)
+	}
+	got, err := s.GetPendingTaskByID(task.ID, "default")
+	if err != nil || got == nil {
+		t.Fatalf("get: %v", err)
+	}
+	if got.Actor != "mcp:codex" || got.ReviewReason != "needs_user" || !got.NeedsReview() {
+		t.Fatalf("task=%+v", got)
+	}
+	if got.ActorLabel() != "Recorded by an AI agent (codex)" {
+		t.Fatalf("label=%q", got.ActorLabel())
+	}
+	human := PendingTask{TaskType: TaskManualForm}
+	if human.ActorLabel() != "" || human.NeedsReview() {
+		t.Fatal("plain form task shown as agent work or needing review")
+	}
+}
+
+func TestUpdateAutomationTaskClearsReviewOnProgress(t *testing.T) {
+	s := newTestStore(t)
+	task, err := s.EnsureAutomationTask("default", "acme", "Acme", TaskManualForm, "https://acme.example/form", "mcp:codex")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.UpdateAutomationTask("default", task.ID, "mcp:codex", "outcome_unknown", "Unknown", "", false); err != nil {
+		t.Fatal(err)
+	}
+	got, _ := s.GetPendingTaskByID(task.ID, "default")
+	if !got.NeedsReview() {
+		t.Fatal("unknown outcome not in review")
+	}
+	if err := s.UpdateAutomationTask("default", task.ID, "mcp:codex", "", "Progress", "", false); err != nil {
+		t.Fatal(err)
+	}
+	got, _ = s.GetPendingTaskByID(task.ID, "default")
+	if got.NeedsReview() {
+		t.Fatal("review stayed after progress")
+	}
+}
+
+func TestPipelineActorIsRecordedAndExported(t *testing.T) {
+	s := newTestStore(t)
+	if err := s.Add(&Record{ProfileID: "default", BrokerID: "acme", BrokerName: "Acme", Status: StatusSent, SentAt: time.Now()}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.UpdatePipelineStatusBy("default", "acme", PipelineAwaitingResponse, "mcp:codex"); err != nil {
+		t.Fatal(err)
+	}
+	latest, _ := s.LatestRequests("default")
+	if latest[0].PipelineStatus != PipelineAwaitingResponse || latest[0].PipelineActor != "mcp:codex" {
+		t.Fatalf("latest=%+v", latest[0])
+	}
+	if err := s.OverridePipelineStatus("default", "acme", PipelineConfirmed); err != nil {
+		t.Fatal(err)
+	}
+	latest, _ = s.LatestRequests("default")
+	if latest[0].PipelineActor != "human" {
+		t.Fatalf("override actor=%q", latest[0].PipelineActor)
+	}
+}

internal/history/automation.go

@@ -100,13 +100,13 @@ func (s *Store) RetryAutomationAction(profileID, key string) error {
 
 // EnsureAutomationTask shows email-derived and catalog forms in the Tasks UI, one
 // task per form however often an agent resumes.
-func (s *Store) EnsureAutomationTask(profileID, brokerID, brokerName string, kind TaskType, rawURL string) (*PendingTask, error) {
+func (s *Store) EnsureAutomationTask(profileID, brokerID, brokerName string, kind TaskType, rawURL, actor string) (*PendingTask, error) {
 	profileID = normalizeProfileID(profileID)
 	_, err := s.db.Exec(`INSERT INTO pending_tasks
-		(profile_id, broker_id, broker_name, task_type, form_url, notes, status, created_at)
-		SELECT ?, ?, ?, ?, ?, 'Opened through MCP', 'pending', ? WHERE NOT EXISTS
+		(profile_id, broker_id, broker_name, task_type, form_url, notes, status, created_at, actor)
+		SELECT ?, ?, ?, ?, ?, 'Opened through MCP', 'pending', ?, ? WHERE NOT EXISTS
 		(SELECT 1 FROM pending_tasks WHERE profile_id = ? AND broker_id = ? AND form_url = ?)`,
-		profileID, brokerID, brokerName, kind, rawURL, time.Now(), profileID, brokerID, rawURL)
+		profileID, brokerID, brokerName, kind, rawURL, time.Now(), actor, profileID, brokerID, rawURL)
 	if err != nil {
 		return nil, err
 	}
@@ -121,14 +121,14 @@ func (s *Store) EnsureAutomationTask(profileID, brokerID, brokerName string, kin
 
 // UpdateAutomationTask writes notes, evidence and completion in one update. A
 // missing, other-profile, completed or skipped task is an error.
-func (s *Store) UpdateAutomationTask(profileID string, id int64, note, screenshot string, completed bool) error {
+func (s *Store) UpdateAutomationTask(profileID string, id int64, actor, reviewReason, note, screenshot string, completed bool) error {
 	status := "pending"
 	var at any
 	if completed {
 		status, at = "completed", time.Now()
 	}
-	r, err := s.db.Exec(`UPDATE pending_tasks SET notes = ?, screenshot_path = ?, status = ?, completed_at = ?
-		WHERE profile_id = ? AND id = ? AND status = 'pending'`, note, screenshot, status, at, normalizeProfileID(profileID), id)
+	r, err := s.db.Exec(`UPDATE pending_tasks SET notes = ?, screenshot_path = ?, status = ?, completed_at = ?, actor = ?, review_reason = ?
+		WHERE profile_id = ? AND id = ? AND status = 'pending'`, note, screenshot, status, at, actor, reviewReason, normalizeProfileID(profileID), id)
 	if err != nil {
 		return err
 	}

internal/history/automation_test.go

@@ -74,25 +74,25 @@ func TestAutomationTaskVisibleAndProfileScoped(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() { _ = s.Close() }()
-	task, err := s.EnsureAutomationTask("alice", "broker", "Broker", TaskManualForm, "https://example.com/form")
+	task, err := s.EnsureAutomationTask("alice", "broker", "Broker", TaskManualForm, "https://example.com/form", "mcp:test")
 	if err != nil {
 		t.Fatal(err)
 	}
-	again, err := s.EnsureAutomationTask("alice", "broker", "Broker", TaskManualForm, "https://example.com/form")
+	again, err := s.EnsureAutomationTask("alice", "broker", "Broker", TaskManualForm, "https://example.com/form", "mcp:test")
 	if err != nil || again.ID != task.ID {
 		t.Fatalf("duplicate task: %v", err)
 	}
-	if err := s.UpdateAutomationTask("bob", task.ID, "accepted", "proof.png", true); err == nil {
+	if err := s.UpdateAutomationTask("bob", task.ID, "mcp:test", "", "accepted", "proof.png", true); err == nil {
 		t.Fatal("cross-profile write succeeded")
 	}
-	if err := s.UpdateAutomationTask("alice", task.ID, "Request accepted; awaiting reply", "proof.png", true); err != nil {
+	if err := s.UpdateAutomationTask("alice", task.ID, "mcp:test", "", "Request accepted; awaiting reply", "proof.png", true); err != nil {
 		t.Fatal(err)
 	}
 	got, err := s.GetPendingTasks("alice", "", "completed")
 	if err != nil || len(got) != 1 || got[0].ScreenshotPath != "proof.png" {
 		t.Fatalf("UI task data: %+v, %v", got, err)
 	}
-	if err := s.UpdateAutomationTask("alice", task.ID, "overwrite", "", false); err == nil {
+	if err := s.UpdateAutomationTask("alice", task.ID, "mcp:test", "", "overwrite", "", false); err == nil {
 		t.Fatal("reopened completed task")
 	}
 }

internal/history/history.go

@@ -67,6 +67,7 @@ type Record struct {
 	SentAt         time.Time
 	CreatedAt      time.Time
 	PipelineStatus PipelineStatus // Current stage in pipeline
+	PipelineActor  string         // "human", "mcp:<client>", or empty for automated updates
 	SentMethod     string         // "smtp" (Eraser sent it) or "manual" (user did)
 }
 
@@ -104,6 +105,10 @@ type PendingTask struct {
 	CreatedAt      time.Time
 	OpenedAt       sql.NullTime // When user first opened the helper page
 	CompletedAt    sql.NullTime
+	// Actor is "human", or "mcp:<client>" for work an agent recorded.
+	Actor string
+	// ReviewReason is set when an automated step needs a person to decide.
+	ReviewReason string
 }
 
 type Store struct {
@@ -204,6 +209,15 @@ func (s *Store) migrate() error {
 	if err := addColumnIfMissing(s.db, `ALTER TABLE removal_requests ADD COLUMN sent_method TEXT DEFAULT 'smtp'`); err != nil {
 		return err
 	}
+	if err := addColumnIfMissing(s.db, `ALTER TABLE pending_tasks ADD COLUMN actor TEXT NOT NULL DEFAULT 'human'`); err != nil {
+		return err
+	}
+	if err := addColumnIfMissing(s.db, `ALTER TABLE pending_tasks ADD COLUMN review_reason TEXT NOT NULL DEFAULT ''`); err != nil {
+		return err
+	}
+	if err := addColumnIfMissing(s.db, `ALTER TABLE removal_requests ADD COLUMN pipeline_actor TEXT NOT NULL DEFAULT ''`); err != nil {
+		return err
+	}
 	if err := addColumnIfMissing(s.db, `ALTER TABLE pending_tasks ADD COLUMN opened_at DATETIME`); err != nil {
 		return err
 	}
@@ -238,6 +252,7 @@ func (s *Store) migrate() error {
 		sent_at DATETIME,
 		created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
 		pipeline_status TEXT DEFAULT 'email_sent',
+		pipeline_actor TEXT NOT NULL DEFAULT '',
 		sent_method TEXT DEFAULT 'smtp'
 	);
 
@@ -290,6 +305,8 @@ func (s *Store) migrate() error {
 		status TEXT DEFAULT 'pending',
 		created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
 		opened_at DATETIME,
+		actor TEXT NOT NULL DEFAULT 'human',
+		review_reason TEXT NOT NULL DEFAULT '',
 		completed_at DATETIME
 	);
 
@@ -383,7 +400,7 @@ func (s *Store) GetRecentRequests(profileID string, limit int) ([]Record, error)
 func (s *Store) GetAllRequests(profileID string) ([]Record, error) {
 	query := `
 	SELECT id, profile_id, broker_id, broker_name, email, template, status,
-	       message_id, error, sent_at, created_at, pipeline_status, sent_method
+	       message_id, error, sent_at, created_at, pipeline_status, sent_method, pipeline_actor
 	FROM removal_requests WHERE profile_id = ? ORDER BY sent_at ASC, id ASC`
 
 	rows, err := s.db.Query(query, normalizeProfileID(profileID))
@@ -396,16 +413,17 @@ func (s *Store) GetAllRequests(profileID string) ([]Record, error) {
 	for rows.Next() {
 		var r Record
 		var sentAt, createdAt sql.NullTime
-		var messageID, errStr, pipelineStatus, sentMethod sql.NullString
+		var messageID, errStr, pipelineStatus, sentMethod, pipelineActor sql.NullString
 		if err := rows.Scan(&r.ID, &r.ProfileID, &r.BrokerID, &r.BrokerName, &r.Email,
-			&r.Template, &r.Status, &messageID, &errStr, &sentAt, &createdAt, &pipelineStatus, &sentMethod); err != nil {
+			&r.Template, &r.Status, &messageID, &errStr, &sentAt, &createdAt, &pipelineStatus, &sentMethod, &pipelineActor); err != nil {
 			return nil, fmt.Errorf("failed to scan request: %w", err)
 		}
 		r.MessageID = messageID.String
 		r.Error = errStr.String
 		r.SentAt = sentAt.Time
 		r.CreatedAt = createdAt.Time
 		r.PipelineStatus = PipelineStatus(pipelineStatus.String)
+		r.PipelineActor = pipelineActor.String
 		r.SentMethod = sentMethod.String
 		records = append(records, r)
 	}
@@ -1034,13 +1052,16 @@ func (s *Store) AddPendingTask(task *PendingTask) error {
 
 	query := `
 	INSERT INTO pending_tasks (profile_id, broker_id, broker_name, task_type, form_url, screenshot_path,
-		browser_state, notes, status, created_at)
-	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		browser_state, notes, status, created_at, actor, review_reason)
+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 	`
+	if task.Actor == "" {
+		task.Actor = "human"
+	}
 
 	result, err := s.db.Exec(query,
 		task.ProfileID, task.BrokerID, task.BrokerName, task.TaskType, task.FormURL, task.ScreenshotPath,
-		task.BrowserState, task.Notes, "pending", time.Now(),
+		task.BrowserState, task.Notes, "pending", time.Now(), task.Actor, task.ReviewReason,
 	)
 	if err != nil {
 		return fmt.Errorf("failed to insert pending task: %w", err)
@@ -1062,22 +1083,22 @@ func (s *Store) GetPendingTasks(profileID string, taskType TaskType, status stri
 
 	if taskType != "" && status != "" {
 		query = `SELECT id, profile_id, broker_id, broker_name, task_type, form_url, screenshot_path,
-			browser_state, notes, status, created_at, opened_at, completed_at
+			browser_state, notes, status, created_at, opened_at, completed_at, actor, review_reason
 			FROM pending_tasks WHERE profile_id = ? AND task_type = ? AND status = ? ORDER BY created_at DESC`
 		args = []interface{}{profileID, taskType, status}
 	} else if taskType != "" {
 		query = `SELECT id, profile_id, broker_id, broker_name, task_type, form_url, screenshot_path,
-			browser_state, notes, status, created_at, opened_at, completed_at
+			browser_state, notes, status, created_at, opened_at, completed_at, actor, review_reason
 			FROM pending_tasks WHERE profile_id = ? AND task_type = ? ORDER BY created_at DESC`
 		args = []interface{}{profileID, taskType}
 	} else if status != "" {
 		query = `SELECT id, profile_id, broker_id, broker_name, task_type, form_url, screenshot_path,
-			browser_state, notes, status, created_at, opened_at, completed_at
+			browser_state, notes, status, created_at, opened_at, completed_at, actor, review_reason
 			FROM pending_tasks WHERE profile_id = ? AND status = ? ORDER BY created_at DESC`
 		args = []interface{}{profileID, status}
 	} else {
 		query = `SELECT id, profile_id, broker_id, broker_name, task_type, form_url, screenshot_path,
-			browser_state, notes, status, created_at, opened_at, completed_at
+			browser_state, notes, status, created_at, opened_at, completed_at, actor, review_reason
 			FROM pending_tasks WHERE profile_id = ? ORDER BY created_at DESC`
 		args = []interface{}{profileID}
 	}
@@ -1095,7 +1116,7 @@ func (s *Store) GetPendingTasks(profileID string, taskType TaskType, status stri
 		var formURL, screenshotPath, browserState, notes sql.NullString
 
 		err := rows.Scan(&t.ID, &t.ProfileID, &t.BrokerID, &t.BrokerName, &t.TaskType, &formURL, &screenshotPath,
-			&browserState, &notes, &t.Status, &createdAt, &t.OpenedAt, &t.CompletedAt)
+			&browserState, &notes, &t.Status, &createdAt, &t.OpenedAt, &t.CompletedAt, &t.Actor, &t.ReviewReason)
 		if err != nil {
 			return nil, fmt.Errorf("failed to scan pending task: %w", err)
 		}
@@ -1116,15 +1137,15 @@ func (s *Store) GetPendingTasks(profileID string, taskType TaskType, status stri
 // returns nil, nil, the same as a task that doesn't exist at all.
 func (s *Store) GetPendingTaskByID(id int64, profileID string) (*PendingTask, error) {
 	query := `SELECT id, profile_id, broker_id, broker_name, task_type, form_url, screenshot_path,
-		browser_state, notes, status, created_at, opened_at, completed_at
+		browser_state, notes, status, created_at, opened_at, completed_at, actor, review_reason
 		FROM pending_tasks WHERE id = ? AND profile_id = ?`
 
 	var t PendingTask
 	var createdAt sql.NullTime
 	var formURL, screenshotPath, browserState, notes sql.NullString
 
 	err := s.db.QueryRow(query, id, normalizeProfileID(profileID)).Scan(&t.ID, &t.ProfileID, &t.BrokerID, &t.BrokerName, &t.TaskType, &formURL, &screenshotPath,
-		&browserState, &notes, &t.Status, &createdAt, &t.OpenedAt, &t.CompletedAt)
+		&browserState, &notes, &t.Status, &createdAt, &t.OpenedAt, &t.CompletedAt, &t.Actor, &t.ReviewReason)
 	if err == sql.ErrNoRows {
 		return nil, nil
 	}

internal/history/pipeline.go

@@ -39,7 +39,18 @@ func (a ActionItems) FormsToFill() int {
 // NeedsReview: review tasks, plus automation outcomes the agent couldn't settle
 // (they stay pending until a person looks).
 func (t PendingTask) NeedsReview() bool {
-	return t.TaskType == TaskReview || strings.HasPrefix(t.Notes, "MCP: unknown.") || strings.HasPrefix(t.Notes, "MCP: needs_user.") || strings.HasPrefix(t.Notes, "MCP: action outcome unknown.") || strings.HasPrefix(t.Notes, "MCP: page could not be opened.")
+	return t.TaskType == TaskReview || t.ReviewReason != ""
+}
+
+// ActorLabel names the AI agent that recorded a task. Tasks you made are unlabelled.
+func (t PendingTask) ActorLabel() string {
+	if name, ok := strings.CutPrefix(t.Actor, "mcp:"); ok {
+		if name == "" {
+			return "Recorded by an AI agent"
+		}
+		return "Recorded by an AI agent (" + name + ")"
+	}
+	return ""
 }
 
 func (s *Store) GetActionItems(profileID string) (ActionItems, error) {