fix: scope completion idempotency to its workflow

internal/workflow/store.go

@@ -276,7 +276,7 @@ func (s *Store) Complete(id, stageID, account, operation string, b store.Object)
 	if current == nil {
 		return nil, fail(404, "工序不存在")
 	}
-	prior, e := query(tx, `SELECT * FROM stage_completions WHERE operation_id=?`, operation)
+	prior, e := query(tx, `SELECT * FROM stage_completions WHERE workflow_id=? AND operation_id=?`, id, operation)
 	if e != nil {
 		return nil, e
 	}

internal/workflow/store_test.go

@@ -68,3 +68,60 @@ func TestDispatchOwnershipOrderingAndAtomicOutbox(t *testing.T) {
 		t.Fatal("outbox mark failed", e)
 	}
 }
+
+func TestCompletionOperationOwnershipAndOutboxRollback(t *testing.T) {
+	s, err := Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.DB.Close()
+	create := func(account, dispatch string) string {
+		result, err := s.Receive(store.Object{"schema_version": "fasttask-dispatch-v1", "dispatch_id": dispatch, "dispatch_revision": float64(1), "receiver_ref": account, "title": "Research", "stages": []any{store.Object{"stage_id": "read", "module": "fast-read", "title": "Read"}}}, account)
+		if err != nil {
+			t.Fatal(err)
+		}
+		id := store.String(result["workflow"].(store.Object)["id"])
+		if _, err := s.Start(id, account); err != nil {
+			t.Fatal(err)
+		}
+		return id
+	}
+	for _, item := range []struct{ dispatch, account string }{{"dispatch-a", "account-a"}, {"dispatch-b", "account-b"}, {"dispatch-c", "account-a"}} {
+		id := create(item.account, item.dispatch)
+		result, err := s.Complete(id, "read", item.account, "shared-operation", store.Object{})
+		if err != nil || result["duplicate"] != false {
+			t.Fatalf("independent completion: %v %v", result, err)
+		}
+		duplicate, err := s.Complete(id, "read", item.account, "shared-operation", store.Object{})
+		if err != nil || duplicate["duplicate"] != true {
+			t.Fatalf("duplicate: %v %v", duplicate, err)
+		}
+	}
+	if events, err := s.Pending(); err != nil || len(events) != 3 {
+		t.Fatalf("events: %v %v", events, err)
+	}
+	rollback := create("account-a", "dispatch-rollback")
+	if _, err := s.DB.Exec(`CREATE TRIGGER reject_outbox BEFORE INSERT ON outbound_events BEGIN SELECT RAISE(ABORT, 'outbox unavailable'); END`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Complete(rollback, "read", "account-a", "retry-operation", store.Object{}); err == nil {
+		t.Fatal("outbox failure accepted")
+	}
+	view, err := s.Get(rollback, "account-a")
+	if err != nil || view["status"] != "active" {
+		t.Fatalf("workflow not rolled back: %v %v", view, err)
+	}
+	var completions int
+	if err := s.DB.QueryRow(`SELECT count(*) FROM stage_completions WHERE workflow_id=?`, rollback).Scan(&completions); err != nil || completions != 0 {
+		t.Fatalf("completion not rolled back: %d %v", completions, err)
+	}
+	if _, err := s.DB.Exec(`DROP TRIGGER reject_outbox`); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.Complete(rollback, "read", "account-a", "retry-operation", store.Object{}); err != nil || result["duplicate"] != false {
+		t.Fatalf("recovery: %v %v", result, err)
+	}
+	if events, err := s.Pending(); err != nil || len(events) != 4 {
+		t.Fatalf("recovered events: %v %v", events, err)
+	}
+}