diff --git a/binlog_coordinate.go b/binlog_coordinate.go
index 00728064..0d968ec4 100644
--- a/binlog_coordinate.go
+++ b/binlog_coordinate.go
@@ -28,14 +28,19 @@ const (
 
 // BinlogCoordinate is a representation-agnostic replication coordinate.
 //
-// Today it only wraps a file/position (mysql.Position). It exists so that the
-// rest of Ghostferry can be migrated to talk in terms of "a coordinate" rather
-// than "a file and a position", which is a prerequisite for adding a GTID mode
-// behind a feature flag without a second invasive refactor.
+// It can express either a file/position (mysql.Position) or a GTID set. It
+// exists so that the rest of Ghostferry can be migrated to talk in terms of "a
+// coordinate" rather than "a file and a position", which is a prerequisite for
+// adding a GTID mode behind a feature flag without a second invasive refactor.
 //
 // The zero value is a zero file/position coordinate, matching the previous
 // behavior where an empty mysql.Position was used as the "no coordinate"
 // sentinel.
+//
+// GTID coordinates are stored canonically as the GTID set string. The parsed
+// mysql.GTIDSet is derived on demand; it is intentionally not stored so that
+// the value type stays trivially copyable and comparable-by-value-free (GTID
+// sets are mutable and must be cloned before mutation).
 type BinlogCoordinate struct {
 	// Type selects the active representation. An empty Type is treated as
 	// BinlogCoordinateFilePosition for backwards compatibility.
@@ -43,6 +48,20 @@ type BinlogCoordinate struct {
 
 	// FilePosition holds the coordinate when Type is BinlogCoordinateFilePosition.
 	FilePosition mysql.Position
+
+	// GTIDSet holds the coordinate as a canonical MySQL GTID set string when
+	// Type is BinlogCoordinateGTID, e.g.
+	// "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-57". An empty string means "no
+	// GTIDs" which is distinct from a nil/unset coordinate; callers that need
+	// that distinction should check Type and IsZero together.
+	GTIDSet string
+
+	// parsedGTIDSet is an optional cache of the parsed GTIDSet. It is populated
+	// by NewGTIDCoordinateFromSet (and lazily by parsedSet) so that hot paths
+	// such as the stop-condition check do not re-parse the canonical string on
+	// every binlog event. It is never mutated in place; accessors that hand a
+	// set to callers clone it. It is intentionally excluded from JSON.
+	parsedGTIDSet mysql.GTIDSet
 }
 
 // NewFilePositionCoordinate wraps a mysql.Position in a BinlogCoordinate.
@@ -53,6 +72,33 @@ func NewFilePositionCoordinate(pos mysql.Position) BinlogCoordinate {
 	}
 }
 
+// NewGTIDCoordinate builds a GTID coordinate from a canonical GTID set string.
+// The string is not validated here; use ParsedGTIDSet or the DB read helpers
+// when a parsed/validated set is required.
+func NewGTIDCoordinate(gtidSet string) BinlogCoordinate {
+	return BinlogCoordinate{
+		Type:    BinlogCoordinateGTID,
+		GTIDSet: gtidSet,
+	}
+}
+
+// NewGTIDCoordinateFromSet builds a GTID coordinate from an already-parsed
+// mysql.GTIDSet. The set is cloned so the coordinate does not alias (or later
+// mutate) the caller's set, and the clone is cached to avoid re-parsing on hot
+// paths. Prefer this over NewGTIDCoordinate when a parsed set is already in
+// hand (e.g. from go-mysql event tracking).
+func NewGTIDCoordinateFromSet(set mysql.GTIDSet) BinlogCoordinate {
+	if set == nil {
+		return BinlogCoordinate{Type: BinlogCoordinateGTID}
+	}
+	cloned := set.Clone()
+	return BinlogCoordinate{
+		Type:          BinlogCoordinateGTID,
+		GTIDSet:       cloned.String(),
+		parsedGTIDSet: cloned,
+	}
+}
+
 // resolvedType returns the effective type, treating the empty string as
 // file/position for backwards compatibility with older serialized state and
 // zero-valued coordinates.
@@ -68,6 +114,11 @@ func (c BinlogCoordinate) IsFilePosition() bool {
 	return c.resolvedType() == BinlogCoordinateFilePosition
 }
 
+// IsGTID reports whether this coordinate is a GTID coordinate.
+func (c BinlogCoordinate) IsGTID() bool {
+	return c.resolvedType() == BinlogCoordinateGTID
+}
+
 // Position returns the underlying mysql.Position.
 //
 // It is valid to call this only for file/position coordinates. This accessor
@@ -78,37 +129,87 @@ func (c BinlogCoordinate) Position() mysql.Position {
 	return c.FilePosition
 }
 
+// ParsedGTIDSet parses and returns the underlying GTID set.
+//
+// It is valid to call this only for GTID coordinates. A fresh mysql.GTIDSet is
+// returned on each call so callers may mutate it freely without affecting the
+// coordinate (the internal cache, if any, is cloned before returning).
+func (c BinlogCoordinate) ParsedGTIDSet() (mysql.GTIDSet, error) {
+	set, err := c.parsedSet()
+	if err != nil {
+		return nil, err
+	}
+	return set.Clone(), nil
+}
+
+// parsedSet returns the parsed GTID set, using the cache when present. The
+// returned set MUST NOT be mutated by callers; use ParsedGTIDSet for a
+// mutable clone. This exists so hot paths (HasReached) avoid re-parsing.
+func (c BinlogCoordinate) parsedSet() (mysql.GTIDSet, error) {
+	if c.resolvedType() != BinlogCoordinateGTID {
+		return nil, fmt.Errorf("parsedSet called on non-GTID coordinate of type %q", c.resolvedType())
+	}
+	if c.parsedGTIDSet != nil {
+		return c.parsedGTIDSet, nil
+	}
+	return mysql.ParseMysqlGTIDSet(c.GTIDSet)
+}
+
 // IsZero reports whether the coordinate carries no meaningful position. This
 // mirrors the previous use of an empty mysql.Position as the "unset" sentinel.
+//
+// For GTID coordinates, an empty GTID set string is considered zero. Note that
+// an empty GTID set is semantically "earlier than everything"; callers that
+// must distinguish "no usable coordinate" from "genuinely empty GTID set"
+// should track that distinction separately (e.g. a nil coordinate pointer).
 func (c BinlogCoordinate) IsZero() bool {
 	switch c.resolvedType() {
 	case BinlogCoordinateFilePosition:
 		return c.FilePosition == (mysql.Position{})
+	case BinlogCoordinateGTID:
+		return c.GTIDSet == ""
 	default:
 		return false
 	}
 }
 
-// Compare orders two coordinates of the same type.
+// HasReached reports whether this coordinate has reached or passed target,
+// i.e. whether a stream positioned at c has already covered everything up to
+// target. This is the single "have we crossed the finish line?" question used
+// for stop and catchup conditions; it hides the representation-specific
+// mechanics from callers.
+//
+//   - For file/position coordinates it is a position comparison (c >= target).
+//   - For GTID coordinates it is set containment (c's set contains target's
+//     set), because GTID sets do not form a total order.
 //
-// It returns -1, 0 or 1 following the semantics of mysql.Position.Compare for
-// file/position coordinates. Comparing coordinates of differing types is a
-// programmer error and panics, because there is no meaningful total ordering
-// across representations. Callers that may receive mixed types should branch on
-// the coordinate type first.
-func (c BinlogCoordinate) Compare(other BinlogCoordinate) int {
-	if c.resolvedType() != other.resolvedType() {
-		panic(fmt.Sprintf(
+// Both coordinates must be the same type; a mismatch returns an error rather
+// than guessing across representations.
+func (c BinlogCoordinate) HasReached(target BinlogCoordinate) (bool, error) {
+	if c.resolvedType() != target.resolvedType() {
+		return false, fmt.Errorf(
 			"cannot compare binlog coordinates of different types: %q vs %q",
-			c.resolvedType(), other.resolvedType(),
-		))
+			c.resolvedType(), target.resolvedType(),
+		)
 	}
 
 	switch c.resolvedType() {
 	case BinlogCoordinateFilePosition:
-		return c.FilePosition.Compare(other.FilePosition)
+		return c.FilePosition.Compare(target.FilePosition) >= 0, nil
+	case BinlogCoordinateGTID:
+		// Use the cached parsed sets (read-only) to avoid re-parsing on hot
+		// paths such as the per-event stop check. Contain does not mutate.
+		mine, err := c.parsedSet()
+		if err != nil {
+			return false, fmt.Errorf("parsing GTID set %q: %w", c.GTIDSet, err)
+		}
+		theirs, err := target.parsedSet()
+		if err != nil {
+			return false, fmt.Errorf("parsing GTID set %q: %w", target.GTIDSet, err)
+		}
+		return mine.Contain(theirs), nil
 	default:
-		panic(fmt.Sprintf("comparison not implemented for binlog coordinate type %q", c.resolvedType()))
+		return false, fmt.Errorf("HasReached not supported for binlog coordinate type %q", c.resolvedType())
 	}
 }
 
@@ -117,6 +218,8 @@ func (c BinlogCoordinate) String() string {
 	switch c.resolvedType() {
 	case BinlogCoordinateFilePosition:
 		return fmt.Sprintf("%s:%d", c.FilePosition.Name, c.FilePosition.Pos)
+	case BinlogCoordinateGTID:
+		return fmt.Sprintf("gtid:%s", c.GTIDSet)
 	default:
 		return fmt.Sprintf("<unknown binlog coordinate type %q>", c.resolvedType())
 	}
@@ -129,6 +232,7 @@ func (c BinlogCoordinate) String() string {
 type serializedBinlogCoordinate struct {
 	Type         BinlogCoordinateType `json:"Type"`
 	FilePosition *mysql.Position      `json:"FilePosition,omitempty"`
+	GTIDSet      string               `json:"GTIDSet,omitempty"`
 }
 
 // MarshalJSON implements json.Marshaler.
@@ -139,6 +243,8 @@ func (c BinlogCoordinate) MarshalJSON() ([]byte, error) {
 	case BinlogCoordinateFilePosition:
 		pos := c.FilePosition
 		out.FilePosition = &pos
+	case BinlogCoordinateGTID:
+		out.GTIDSet = c.GTIDSet
 	default:
 		return nil, fmt.Errorf("cannot marshal binlog coordinate of type %q", c.resolvedType())
 	}
@@ -162,6 +268,9 @@ func (c *BinlogCoordinate) UnmarshalJSON(data []byte) error {
 				c.FilePosition = mysql.Position{}
 			}
 			return nil
+		case BinlogCoordinateGTID:
+			c.GTIDSet = typed.GTIDSet
+			return nil
 		default:
 			return fmt.Errorf("cannot unmarshal binlog coordinate of type %q", typed.Type)
 		}
diff --git a/config.go b/config.go
index d696b79a..afbcc41e 100644
--- a/config.go
+++ b/config.go
@@ -684,6 +684,22 @@ type Config struct {
 	// Make sure you have binlog_row_image=FULL when turning on this
 	SkipBinlogRowImageCheck bool
 
+	// BinlogCoordinateMode selects how Ghostferry expresses and persists binlog
+	// coordinates.
+	//
+	// Valid values:
+	//   - "" or "file_position": the classic (file, position) coordinate. This
+	//     is the default and current behavior.
+	//   - "gtid": GTID-set based coordinates (experimental, MySQL only). At this
+	//     stage GTID mode only affects coordinate representation, validation, and
+	//     the ability to read the current GTID set; binlog streaming still uses
+	//     file/position until a later stage wires GTID streaming.
+	//
+	// Prefer BinlogCoordinateMode over any future boolean flag: it keeps the
+	// file/position and GTID paths cleanly separated and leaves room for
+	// additional coordinate sources.
+	BinlogCoordinateMode BinlogCoordinateType
+
 	// This config is necessary for inline verification for a special case of
 	// Ghostferry:
 	//
@@ -853,6 +869,15 @@ func (c *Config) ValidateConfig() error {
 		return fmt.Errorf("StateToResumeFrom version mismatch: resume = %s, current = %s", c.StateToResumeFrom.GhostferryVersion, VersionString)
 	}
 
+	switch c.BinlogCoordinateMode {
+	case "":
+		c.BinlogCoordinateMode = BinlogCoordinateFilePosition
+	case BinlogCoordinateFilePosition, BinlogCoordinateGTID:
+		// valid
+	default:
+		return fmt.Errorf("invalid BinlogCoordinateMode %q, must be %q or %q", c.BinlogCoordinateMode, BinlogCoordinateFilePosition, BinlogCoordinateGTID)
+	}
+
 	if c.VerifierType == VerifierTypeIterative {
 		if err := c.IterativeVerifierConfig.Validate(); err != nil {
 			return fmt.Errorf("IterativeVerifierConfig invalid: %v", err)
diff --git a/test/go/binlog_coordinate_test.go b/test/go/binlog_coordinate_test.go
index 808e50fa..e732de22 100644
--- a/test/go/binlog_coordinate_test.go
+++ b/test/go/binlog_coordinate_test.go
@@ -28,15 +28,23 @@ func TestBinlogCoordinate_ZeroValueIsFilePosition(t *testing.T) {
 	assert.Equal(t, mysql.Position{}, coord.Position())
 }
 
-func TestBinlogCoordinate_Compare(t *testing.T) {
+func TestBinlogCoordinate_HasReachedFilePosition(t *testing.T) {
 	a := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 10})
 	b := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 20})
-	c := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000002", Pos: 5})
 
-	assert.Equal(t, -1, a.Compare(b))
-	assert.Equal(t, 1, b.Compare(a))
-	assert.Equal(t, 0, a.Compare(a))
-	assert.Equal(t, -1, b.Compare(c))
+	// b is ahead of a: b has reached a, a has not reached b.
+	reached, err := b.HasReached(a)
+	require.NoError(t, err)
+	assert.True(t, reached)
+
+	reached, err = a.HasReached(b)
+	require.NoError(t, err)
+	assert.False(t, reached)
+
+	// A coordinate has reached itself (>=).
+	reached, err = a.HasReached(a)
+	require.NoError(t, err)
+	assert.True(t, reached)
 }
 
 func TestBinlogCoordinate_JSONRoundTrip(t *testing.T) {
@@ -76,3 +84,105 @@ func TestBinlogCoordinate_UnmarshalTypedForm(t *testing.T) {
 	assert.Equal(t, "mysql-bin.000050", decoded.Position().Name)
 	assert.Equal(t, uint32(700), decoded.Position().Pos)
 }
+
+const (
+	testGTIDSetA = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-57"
+	testGTIDSetB = "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-100"
+)
+
+func TestBinlogCoordinate_GTIDBasics(t *testing.T) {
+	coord := ghostferry.NewGTIDCoordinate(testGTIDSetA)
+
+	assert.True(t, coord.IsGTID())
+	assert.False(t, coord.IsFilePosition())
+	assert.False(t, coord.IsZero())
+	assert.Equal(t, "gtid:"+testGTIDSetA, coord.String())
+
+	set, err := coord.ParsedGTIDSet()
+	require.NoError(t, err)
+	assert.Equal(t, testGTIDSetA, set.String())
+}
+
+func TestBinlogCoordinate_GTIDEmptyIsZero(t *testing.T) {
+	coord := ghostferry.NewGTIDCoordinate("")
+
+	assert.True(t, coord.IsGTID())
+	assert.True(t, coord.IsZero())
+}
+
+func TestBinlogCoordinate_NewGTIDCoordinateFromSet(t *testing.T) {
+	set, err := mysql.ParseMysqlGTIDSet(testGTIDSetA)
+	require.NoError(t, err)
+
+	coord := ghostferry.NewGTIDCoordinateFromSet(set)
+	assert.True(t, coord.IsGTID())
+	assert.Equal(t, testGTIDSetA, coord.GTIDSet)
+
+	// Mutating the caller's set must not affect the coordinate (it clones).
+	other, err := mysql.ParseMysqlGTIDSet(testGTIDSetB)
+	require.NoError(t, err)
+	require.NoError(t, set.(*mysql.MysqlGTIDSet).Add(*other.(*mysql.MysqlGTIDSet)))
+	assert.Equal(t, testGTIDSetA, coord.GTIDSet, "coordinate must not alias the caller's set")
+
+	// The parsed-set cache is not serialized; JSON only carries the string.
+	data, err := json.Marshal(coord)
+	require.NoError(t, err)
+	assert.NotContains(t, string(data), "parsedGTIDSet")
+
+	var decoded ghostferry.BinlogCoordinate
+	require.NoError(t, json.Unmarshal(data, &decoded))
+	reached, err := decoded.HasReached(ghostferry.NewGTIDCoordinate(testGTIDSetA))
+	require.NoError(t, err)
+	assert.True(t, reached)
+}
+
+func TestBinlogCoordinate_NewGTIDCoordinateFromNilSet(t *testing.T) {
+	coord := ghostferry.NewGTIDCoordinateFromSet(nil)
+	assert.True(t, coord.IsGTID())
+	assert.True(t, coord.IsZero())
+}
+
+func TestBinlogCoordinate_HasReachedGTID(t *testing.T) {
+	larger := ghostferry.NewGTIDCoordinate(testGTIDSetB)
+	smaller := ghostferry.NewGTIDCoordinate(testGTIDSetA)
+
+	// The larger set contains the smaller, so it has reached it.
+	reached, err := larger.HasReached(smaller)
+	require.NoError(t, err)
+	assert.True(t, reached)
+
+	reached, err = smaller.HasReached(larger)
+	require.NoError(t, err)
+	assert.False(t, reached)
+}
+
+func TestBinlogCoordinate_HasReachedRejectsMixedTypes(t *testing.T) {
+	gtid := ghostferry.NewGTIDCoordinate(testGTIDSetA)
+	filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4})
+
+	_, err := gtid.HasReached(filePos)
+	assert.Error(t, err)
+
+	_, err = filePos.HasReached(gtid)
+	assert.Error(t, err)
+}
+
+func TestBinlogCoordinate_ParsedGTIDSetRejectsFilePosition(t *testing.T) {
+	filePos := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 4})
+
+	_, err := filePos.ParsedGTIDSet()
+	assert.Error(t, err)
+}
+
+func TestBinlogCoordinate_GTIDJSONRoundTrip(t *testing.T) {
+	coord := ghostferry.NewGTIDCoordinate(testGTIDSetA)
+
+	data, err := json.Marshal(coord)
+	require.NoError(t, err)
+
+	var decoded ghostferry.BinlogCoordinate
+	require.NoError(t, json.Unmarshal(data, &decoded))
+
+	assert.True(t, decoded.IsGTID())
+	assert.Equal(t, testGTIDSetA, decoded.GTIDSet)
+}
diff --git a/test/go/config_test.go b/test/go/config_test.go
index 702e9be5..50b8d0f1 100644
--- a/test/go/config_test.go
+++ b/test/go/config_test.go
@@ -201,6 +201,25 @@ func (this *ConfigTestSuite) TestDefaultMarginalia() {
 	this.Require().Equal(ghostferry.DefaultMarginalia, this.config.Target.Marginalia)
 }
 
+func (this *ConfigTestSuite) TestBinlogCoordinateModeDefaultsToFilePosition() {
+	err := this.config.ValidateConfig()
+	this.Require().Nil(err)
+	this.Require().Equal(ghostferry.BinlogCoordinateFilePosition, this.config.BinlogCoordinateMode)
+}
+
+func (this *ConfigTestSuite) TestBinlogCoordinateModeGTIDIsValid() {
+	this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID
+	err := this.config.ValidateConfig()
+	this.Require().Nil(err)
+	this.Require().Equal(ghostferry.BinlogCoordinateGTID, this.config.BinlogCoordinateMode)
+}
+
+func (this *ConfigTestSuite) TestBinlogCoordinateModeInvalidIsRejected() {
+	this.config.BinlogCoordinateMode = "banana"
+	err := this.config.ValidateConfig()
+	this.Require().EqualError(err, `invalid BinlogCoordinateMode "banana", must be "file_position" or "gtid"`)
+}
+
 func TestConfig(t *testing.T) {
 	testhelpers.SetupTest()
 	suite.Run(t, new(ConfigTestSuite))
diff --git a/utils.go b/utils.go
index 12ee253e..57caba89 100644
--- a/utils.go
+++ b/utils.go
@@ -258,6 +258,54 @@ func CheckDbIsAReplica(db *sql.DB) (bool, error) {
 	return isReadOnly, err
 }
 
+// ReadExecutedGTIDSet reads @@GLOBAL.GTID_EXECUTED from the given database and
+// returns it as a canonical GTID set string. The returned string may be empty
+// if the server has executed no GTID-tagged transactions yet; that is not an
+// error.
+func ReadExecutedGTIDSet(db *sql.DB) (string, error) {
+	var gtidSet string
+	row := db.QueryRow("SELECT @@GLOBAL.GTID_EXECUTED")
+	if err := row.Scan(&gtidSet); err != nil {
+		return "", fmt.Errorf("reading @@GLOBAL.GTID_EXECUTED: %w", err)
+	}
+	return gtidSet, nil
+}
+
+// ReadCurrentGTIDCoordinate reads the current executed GTID set and returns it
+// as a GTID BinlogCoordinate. The raw string is validated by parsing it.
+func ReadCurrentGTIDCoordinate(db *sql.DB) (BinlogCoordinate, error) {
+	gtidSet, err := ReadExecutedGTIDSet(db)
+	if err != nil {
+		return BinlogCoordinate{}, err
+	}
+
+	// Validate that the string parses as a MySQL GTID set. An empty string is
+	// a valid (empty) set.
+	if _, err := mysql.ParseMysqlGTIDSet(gtidSet); err != nil {
+		return BinlogCoordinate{}, fmt.Errorf("parsing executed GTID set %q: %w", gtidSet, err)
+	}
+
+	return NewGTIDCoordinate(gtidSet), nil
+}
+
+// CheckServerGTIDModeEnabled verifies that the server has GTID mode enabled
+// (@@GLOBAL.gtid_mode = ON). It returns an error describing the current value
+// otherwise. This is intended to be called during startup validation when
+// BinlogCoordinateMode is "gtid".
+func CheckServerGTIDModeEnabled(db *sql.DB) error {
+	var gtidMode string
+	row := db.QueryRow("SELECT @@GLOBAL.gtid_mode")
+	if err := row.Scan(&gtidMode); err != nil {
+		return fmt.Errorf("reading @@GLOBAL.gtid_mode: %w", err)
+	}
+
+	if !strings.EqualFold(gtidMode, "ON") {
+		return fmt.Errorf("gtid_mode must be ON for GTID binlog coordinate mode, but was %q", gtidMode)
+	}
+
+	return nil
+}
+
 func ConvertTableColumnsToStrings(columns []schema.TableColumn) []string {
 	out := make([]string, 0, len(columns))
 	for _, column := range columns {