diff --git a/dgraph/cmd/alpha/txn_test.go b/dgraph/cmd/alpha/txn_test.go
index 54e6bc11582..c95372e7a58 100644
--- a/dgraph/cmd/alpha/txn_test.go
+++ b/dgraph/cmd/alpha/txn_test.go
@@ -21,6 +21,11 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 
 	"github.com/dgraph-io/dgo/v250"
 	"github.com/dgraph-io/dgo/v250/protos/api"
@@ -248,6 +253,117 @@ func TestConflict(t *testing.T) {
 	require.True(t, bytes.Equal(resp.Json, []byte("{\"me\":[{\"name\":\"Manish\"}]}")))
 }
 
+// TestConflictAbortReason proves the server emits the categorized abort reason on the
+// gRPC status of a write-write conflict: the code stays codes.Aborted (so existing
+// clients keep retrying) and the message is prefixed with "conflict: ". This is exactly
+// the unflattened status a gRPC client such as dgraph4j receives and parses into
+// TxnConflictException.AbortReason.
+//
+// It must inspect the raw CommitOrAbort response rather than dgo's high-level
+// Txn.Commit, because dgo intentionally replaces any codes.Aborted error with the
+// static dgo.ErrAborted (txn.go), discarding the reason for the Go client.
+func TestConflictAbortReason(t *testing.T) {
+	op := &api.Operation{}
+	op.DropAll = true
+	require.NoError(t, dg.Alter(context.Background(), op))
+
+	// First transaction creates a node with a name.
+	txn := dg.NewTxn()
+	mu := &api.Mutation{}
+	mu.SetJson = []byte(`{"name": "Manish"}`)
+	assigned, err := txn.Mutate(context.Background(), mu)
+	require.NoError(t, err)
+	require.Len(t, assigned.Uids, 1)
+	var uid string
+	for _, u := range assigned.Uids {
+		uid = u
+	}
+
+	// Second transaction writes the same predicate on the same uid -> conflicts.
+	txn2 := dg.NewTxn()
+	mu = &api.Mutation{}
+	mu.SetJson = []byte(fmt.Sprintf(`{"uid": %q, "name": "Manish"}`, uid))
+	resp2, err := txn2.Mutate(context.Background(), mu)
+	require.NoError(t, err)
+	require.NotEmpty(t, resp2.GetTxn().GetKeys(), "mutation response must carry conflict keys")
+
+	// First commit wins; its commitTs is now greater than txn2's startTs.
+	require.NoError(t, txn.Commit(context.Background()))
+
+	// Commit the loser via the raw gRPC stub so we observe the unflattened status the
+	// server sends (dgo's Txn.Commit would replace it with the reasonless ErrAborted).
+	conn, err := grpc.NewClient(alphaSockAdd, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() { _ = conn.Close() }()
+	raw := api.NewDgraphClient(conn)
+
+	// TestMain enables ACL, so the raw stub (unlike the logged-in dg client) must
+	// carry the access JWT; CommitOrAbort goes through the same auth interceptor.
+	ctx := metadata.NewOutgoingContext(context.Background(),
+		metadata.Pairs("accessJwt", hc.AccessJwt))
+	_, err = raw.CommitOrAbort(ctx, resp2.GetTxn())
+	require.Error(t, err)
+
+	st := status.Convert(err)
+	require.Equal(t, codes.Aborted, st.Code(),
+		"abort must keep codes.Aborted so existing clients still retry")
+	require.True(t, strings.HasPrefix(st.Message(), "conflict: "),
+		"abort reason should be categorized as conflict; got %q", st.Message())
+}
+
+// TestPredicateMoveAbortReason proves the "predicate-move" category is surfaced on the
+// gRPC status. Zero's checkPreds aborts a commit whose predicate keys don't match the
+// tablet's serving group (the same code path that rejects commits during a predicate
+// move). We reach it deterministically by committing a real transaction context whose Preds have
+// been rewritten to claim a group that doesn't serve the predicate, with conflict Keys
+// cleared so hasConflict passes and checkPreds runs.
+//
+// As with TestConflictAbortReason, it uses the raw CommitOrAbort stub to observe the
+// unflattened status the server sends.
+func TestPredicateMoveAbortReason(t *testing.T) {
+	op := &api.Operation{}
+	op.DropAll = true
+	require.NoError(t, dg.Alter(context.Background(), op))
+
+	// A normal mutation gives us a valid, fresh TxnContext (real StartTs and Preds).
+	txn := dg.NewTxn()
+	mu := &api.Mutation{SetJson: []byte(`{"name": "Alice"}`)}
+	resp, err := txn.Mutate(context.Background(), mu)
+	require.NoError(t, err)
+
+	tc := resp.GetTxn()
+	require.NotEmpty(t, tc.GetPreds(), "mutation response must carry predicate keys")
+
+	// Rewrite each predicate key's group id to a group that does not serve it, and drop
+	// the conflict keys so hasConflict is false and the commit reaches checkPreds.
+	doctored := make([]string, 0, len(tc.GetPreds()))
+	for _, p := range tc.GetPreds() {
+		// Preds look like "<gid>-<predicate>"; claim a nonexistent group 100.
+		if idx := strings.IndexByte(p, '-'); idx >= 0 {
+			doctored = append(doctored, "100"+p[idx:])
+		}
+	}
+	require.NotEmpty(t, doctored, "expected parseable predicate keys")
+	tc.Preds = doctored
+	tc.Keys = nil
+
+	conn, err := grpc.NewClient(alphaSockAdd, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() { _ = conn.Close() }()
+	raw := api.NewDgraphClient(conn)
+
+	ctx := metadata.NewOutgoingContext(context.Background(),
+		metadata.Pairs("accessJwt", hc.AccessJwt))
+	_, err = raw.CommitOrAbort(ctx, tc)
+	require.Error(t, err)
+
+	st := status.Convert(err)
+	require.Equal(t, codes.Aborted, st.Code(),
+		"abort must keep codes.Aborted so existing clients still retry")
+	require.True(t, strings.HasPrefix(st.Message(), "predicate-move: "),
+		"abort reason should be categorized as predicate-move; got %q", st.Message())
+}
+
 func TestConflictTimeout(t *testing.T) {
 	var uid string
 	txn := dg.NewTxn()
diff --git a/dgraph/cmd/zero/oracle.go b/dgraph/cmd/zero/oracle.go
index 663f8ded0d4..d169c0560a1 100644
--- a/dgraph/cmd/zero/oracle.go
+++ b/dgraph/cmd/zero/oracle.go
@@ -18,6 +18,8 @@ import (
 	"go.opentelemetry.io/otel"
 	attribute "go.opentelemetry.io/otel/attribute"
 	trace "go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/dgraph-io/badger/v4/y"
 	"github.com/dgraph-io/dgo/v250/protos/api"
@@ -338,56 +340,220 @@ func (s *Server) proposeTxn(ctx context.Context, src *api.TxnContext) error {
 	return nil
 }
 
+// Abort-reason codes. When Zero decides to abort a transaction it surfaces the category to the
+// client as the prefix of a codes.Aborted gRPC status message, formatted as "<code>: <detail>".
+// api.TxnContext (external dgo module) has no field for the reason, so the reason rides on the
+// error instead; the status code stays codes.Aborted, so existing abort handling is unaffected.
+// The dgraph4j and pydgraph clients parse these prefixes into their AbortReason enums — keep all
+// three in sync.
+//
+// These three codes are the entire published vocabulary. A cause that cannot be mapped onto one of
+// them emits the detail with *no* prefix (abortReasonUncategorized) rather than borrowing a code
+// that would imply the wrong remedy. Both dgraph4j and pydgraph already degrade an absent or
+// unrecognized prefix to UNKNOWN, so "no category" is a supported and truthful answer. Withholding
+// also keeps a later phase purely additive: naming a new category then only adds precision to
+// something that was already UNKNOWN, and never reclassifies a code clients have shipped against.
+const (
+	// abortReasonUncategorized means the server knows what happened and says so in the detail,
+	// but the cause does not correspond to any published category. It is not "the server does not
+	// know" — it is "there is no code for this yet".
+	abortReasonUncategorized = ""
+	abortReasonConflict      = "conflict"
+	abortReasonStaleStartTs  = "stale-startts"
+	abortReasonPredicateMove = "predicate-move"
+)
+
+// Human-readable details, one per cause. Every message a client can receive for a server-decided
+// abort is declared here rather than at its call site, so the published surface is a single block:
+// adding a new abort means adding a constant next to the existing vocabulary, where the category it
+// should pair with is visible. Entries ending in Fmt are format strings and must be given their
+// arguments; the other two are complete messages.
+//
+//	category         detail
+//	───────────────  ──────────────────────────────────────────────────────────
+//	conflict         abortDetailConflict
+//	stale-startts    abortDetailStaleStartTs
+//	predicate-move   abortDetailPredicateBlockedFmt, abortDetailGroupMismatchFmt
+//	(withheld)       abortDetailMissingGroupIDFmt, abortDetailBadGroupIDFmt,
+//	                 abortDetailTabletNilFmt, abortDetailPreAborted, and
+//	                 ctx.Err() — the last supplied by the Go runtime, so it is
+//	                 not declared here
+const (
+	// A conflict is the one category the server cannot narrow. Every conflict key reaching Zero is a
+	// farm.Fingerprint64(key)^uid produced by GetConflictKey (posting/list.go), so by the time
+	// hasConflict matches one, which key produced it is already unrecoverable. Naming the
+	// possibilities is therefore the most this category can honestly say, and it is worth saying: a
+	// caller who set a single scalar value has no reason to expect that the index and count keys
+	// *derived* from it can conflict too.
+	//
+	// Exactly three kinds of key carry a conflict — data, index and count. Reverse keys carry none:
+	// IsReverse is a distinct ByteType, so a reverse key matches no case in GetConflictKey's switch,
+	// falls to default, and addConflictKey drops the resulting zero. Do not add "reverse" here.
+	//
+	// @upsert is named separately because it is a *rule*, not a fourth kind of key. HasUpsert is
+	// tested ahead of every IsData/IsIndex case, so on an @upsert predicate the existing data and
+	// index keys switch to getKey(key, 0) — the uid drops out and any two transactions writing the
+	// same value collide on the shared index key. That is the uniqueness mechanism, and in
+	// upsert-heavy ingest it is the most likely cause of this abort, which is why it is worth the
+	// extra sentence.
+	//
+	// The leading sentence is load-bearing and must stay verbatim at the front. Three tests in this
+	// repo substring-match it against the live server message — including the retry loop at
+	// dgraph/cmd/alpha/upsert_test.go, which would exit immediately and fail if it stopped matching
+	// — and it is also the text of dgo.ErrAborted, so user log-scrapers key off it. Extend this
+	// detail only by appending.
+	abortDetailConflict = "Transaction has been aborted. Please retry. Another transaction " +
+		"committed to one of the same keys. The conflicting key cannot be identified: it may be " +
+		"the data key written directly, or an index or count key derived from it. On an @upsert " +
+		"predicate the uid is excluded, so any two transactions writing the same value conflict"
+	// A start timestamp goes stale for two different reasons, so this detail deliberately names
+	// both rather than asserting one. Zero raises startTxnTs either when a Zero becomes leader and
+	// renews its leases (updateLeases, assign.go) or when it trims its conflict map while applying
+	// a snapshot (purgeBelow via applySnapshot, raft.go) — the second is not a leader change at
+	// all. Both mean the same thing to a caller: this transaction's start timestamp is older than
+	// the oldest timestamp Zero can still validate against, so retry with a fresh transaction.
+	abortDetailStaleStartTs = "Transaction start timestamp is older than the oldest timestamp " +
+		"Zero can still validate (Zero leader change, or its conflict map was trimmed at a " +
+		"snapshot). Please retry"
+
+	// Reported when proposeTxn comes back with the transaction already aborted while this commit
+	// decided nothing itself — i.e. something aborted it out of band, before the commit was
+	// arbitrated. Zero records no cause when that happens (TryAbort carries only timestamps), so the
+	// category is withheld and the detail names the possibilities instead of guessing. All three
+	// reach Zero through the same TryAbort RPC, and none of them is a predicate *move*, so
+	// predicate-move would be the wrong label:
+	//
+	//   - a schema or type update on a predicate the transaction touched, which cancels pending
+	//     transactions so the index can be rebuilt (detectPendingTxns, worker/draft.go)
+	//   - a drop-predicate (S * * delete) on a predicate the transaction touched (same function)
+	//   - the Alpha leader ageing out transactions idle longer than --limit "txn-abort-after"
+	//     (abortOldTransactions, worker/draft.go)
+	//
+	// Before this was handled, commit() returned nil here and the abort reached the client as a
+	// bare dgo.ErrAborted with no reason at all.
+	abortDetailPreAborted = "Transaction has been aborted. Please retry. It was already aborted " +
+		"before this commit was decided, which happens when a schema update or a drop-predicate " +
+		"cancels pending transactions on a predicate it touched, or when the server ages out " +
+		"transactions idle for longer than --limit \"txn-abort-after\""
+
+	// Details produced by checkPreds. Wording is unchanged from before the abort-reason work; only
+	// the declaration moved here, so existing log-scrapers and docs still match.
+	abortDetailMissingGroupIDFmt   = "Unable to find group id in %s"
+	abortDetailBadGroupIDFmt       = "unable to parse group id from %s"
+	abortDetailTabletNilFmt        = "Tablet for %s is nil"
+	abortDetailPredicateBlockedFmt = "Commits on predicate %s are blocked due to predicate move"
+	abortDetailGroupMismatchFmt    = "Mutation done in group: %d. Predicate %s assigned to %d"
+)
+
+// abortReason builds the wire string the client parses: "<code>: <detail>". An empty code yields
+// the bare detail, which is exactly what a client receives from a server without this feature.
+func abortReason(code, detail string) string {
+	if code == abortReasonUncategorized {
+		return detail
+	}
+	return code + ": " + detail
+}
+
+// conflictAbortReason returns the wire reason for an abort decided by hasConflict, distinguishing a
+// write-write conflict from a stale start timestamp. See isStaleStartTs for what "stale" means.
+func conflictAbortReason(stale bool) string {
+	if stale {
+		return abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs)
+	}
+	return abortReason(abortReasonConflict, abortDetailConflict)
+}
+
+// isStaleStartTs reports whether this transaction's start timestamp (src.StartTs) is below the
+// floor Zero can still validate against (startTxnTs). hasConflict aborts such a transaction, but
+// the cause is not a write-write conflict, so it is reported under its own category. See
+// abortDetailStaleStartTs for the two ways that floor rises. Callers must hold at least a read
+// lock on the oracle.
+func (o *Oracle) isStaleStartTs(src *api.TxnContext) bool {
+	return src.StartTs < o.startTxnTs
+}
+
+// checkPreds reports whether any predicate this transaction touched is unusable, in which case the
+// transaction must be aborted. It returns the abort category alongside the error because its exits
+// are unrelated causes, not degrees of one cause: only a group mismatch and a blocked tablet are
+// predicate moves. A malformed entry in preds, or a predicate that no group serves, are different
+// failures needing different remedies — retrying a malformed predicate key can never succeed — so
+// those return abortReasonUncategorized rather than claiming a move that did not happen. Every exit
+// still aborts the transaction, exactly as before; only the reported category changes.
+//
+// Each entry in preds is a "<group id>-<predicate name>" string built by the Alpha.
+func (s *Server) checkPreds(preds []string) (string, error) {
+	for _, pkey := range preds {
+		splits := strings.SplitN(pkey, "-", 2)
+		if len(splits) < 2 {
+			return abortReasonUncategorized, errors.Errorf(abortDetailMissingGroupIDFmt, pkey)
+		}
+		gid, err := strconv.Atoi(splits[0])
+		if err != nil {
+			return abortReasonUncategorized, errors.Wrapf(err, abortDetailBadGroupIDFmt, pkey)
+		}
+		pred := splits[1]
+		if strings.Contains(pred, hnsw.VecKeyword) {
+			pred = pred[0:strings.Index(pred, hnsw.VecKeyword)]
+		}
+		// Checked before the tablet lookup because blockTablet holds for the entire duration of a
+		// move: movePredicate defers its unblock past the tablet reassignment proposal, so this is
+		// the authoritative signal that a move is in flight. Checking it first means an in-flight
+		// move can never be reported as anything else, and it lets the tablet == nil case below mean
+		// only "no group serves this predicate" and never "it is moving".
+		if s.isBlocked(pred) {
+			return abortReasonPredicateMove, errors.Errorf(abortDetailPredicateBlockedFmt, pred)
+		}
+		tablet := s.ServingTablet(pred)
+		if tablet == nil {
+			return abortReasonUncategorized, errors.Errorf(abortDetailTabletNilFmt, pred)
+		}
+		// The predicate finished moving to another group while this transaction was open: the
+		// mutation was written against group gid, but the predicate now belongs elsewhere.
+		if tablet.GroupId != uint32(gid) {
+			return abortReasonPredicateMove, errors.Errorf(abortDetailGroupMismatchFmt,
+				gid, pred, tablet.GroupId)
+		}
+
+	}
+	return abortReasonUncategorized, nil
+}
+
 func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	span := trace.SpanFromContext(ctx)
 	span.SetAttributes(attribute.Int64("startTs", int64(src.StartTs)))
 	if src.Aborted {
+		// The client discarded this transaction itself (a Discard call), so the server decided
+		// nothing and there is no reason to report.
 		return s.proposeTxn(ctx, src)
 	}
 
+	// abortWithReason marks the transaction aborted, proposes it (advancing the watermark exactly as
+	// the previous code did), and then surfaces the categorized reason to the caller as a gRPC
+	// Aborted status. proposeTxn still runs identically — only the post-propose return changes.
+	abortWithReason := func(reason string) error {
+		span.SetAttributes(attribute.Bool("abort", true))
+		src.Aborted = true
+		if err := s.proposeTxn(ctx, src); err != nil {
+			return err
+		}
+		return status.Error(codes.Aborted, reason)
+	}
+
 	// Use the start timestamp to check if we have a conflict, before we need to assign a commit ts.
 	s.orc.RLock()
 	conflict := s.orc.hasConflict(src)
+	// hasConflict also aborts a transaction whose start timestamp is too old to validate, which is
+	// not a write-write conflict, so it is reported under its own category. See isStaleStartTs.
+	stale := s.orc.isStaleStartTs(src)
 	s.orc.RUnlock()
 	if conflict {
-		span.SetAttributes(attribute.Bool("abort", true))
-		src.Aborted = true
-		return s.proposeTxn(ctx, src)
+		return abortWithReason(conflictAbortReason(stale))
 	}
 
-	checkPreds := func() error {
-		// Check if any of these tablets is being moved. If so, abort the transaction.
-		for _, pkey := range src.Preds {
-			splits := strings.SplitN(pkey, "-", 2)
-			if len(splits) < 2 {
-				return errors.Errorf("Unable to find group id in %s", pkey)
-			}
-			gid, err := strconv.Atoi(splits[0])
-			if err != nil {
-				return errors.Wrapf(err, "unable to parse group id from %s", pkey)
-			}
-			pred := splits[1]
-			if strings.Contains(pred, hnsw.VecKeyword) {
-				pred = pred[0:strings.Index(pred, hnsw.VecKeyword)]
-			}
-			tablet := s.ServingTablet(pred)
-			if tablet == nil {
-				return errors.Errorf("Tablet for %s is nil", pred)
-			}
-			if tablet.GroupId != uint32(gid) {
-				return errors.Errorf("Mutation done in group: %d. Predicate %s assigned to %d",
-					gid, pred, tablet.GroupId)
-			}
-			if s.isBlocked(pred) {
-				return errors.Errorf("Commits on predicate %s are blocked due to predicate move", pred)
-			}
-		}
-		return nil
-	}
-	if err := checkPreds(); err != nil {
-		span.SetAttributes(attribute.Bool("abort", true))
-		src.Aborted = true
-		return s.proposeTxn(ctx, src)
+	if reason, err := s.checkPreds(src.Preds); err != nil {
+		// checkPreds already builds a specific, human-readable message for each cause. Forward it
+		// instead of discarding it, tagged with whichever category that cause justifies.
+		return abortWithReason(abortReason(reason, err.Error()))
 	}
 
 	num := pb.Num{Val: 1, Type: pb.Num_TXN_TS}
@@ -402,16 +568,65 @@ func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	span.SetAttributes(attribute.Int64("nodeId", int64(s.Node.Id)))
 	span.AddEvent(fmt.Sprintf("TXN Context: %+v", src))
 
+	// Past this point the transaction already holds a commit timestamp, so an abort here is a "late"
+	// abort. Two unrelated causes can trigger one, tracked separately because they report
+	// differently. A conflict found here takes precedence: if the oracle rejected the transaction it
+	// would have aborted whatever the context said, so that is the honest cause to report.
+	aborted := false
+	lateReason := abortReasonUncategorized
+	lateDetail := abortDetailConflict
 	if err := s.orc.commit(src); err != nil {
 		span.SetAttributes(attribute.Bool("abort", true))
 		src.Aborted = true
+		aborted = true
+		// Oracle.commit re-runs hasConflict, which aborts a stale start timestamp just as readily as
+		// a genuine write-write conflict found in the keyCommit tree, and collapses both into
+		// x.ErrConflict. Re-check staleness here so this path can report stale-startts, instead of
+		// labelling every late abort a conflict. The early check at the top of commit() already
+		// draws that distinction; without this, the stale-startts category was only ever reachable
+		// from the early path.
+		s.orc.RLock()
+		stale := s.orc.isStaleStartTs(src)
+		s.orc.RUnlock()
+		lateReason = abortReasonConflict
+		if stale {
+			lateReason = abortReasonStaleStartTs
+			lateDetail = abortDetailStaleStartTs
+		}
 	}
 	if err := ctx.Err(); err != nil {
 		span.SetAttributes(attribute.Bool("abort", true))
 		src.Aborted = true
+		if !aborted {
+			// The context was cancelled or timed out; no conflict occurred. No published category
+			// describes this, so report the context's own message with no category prefix rather
+			// than claiming a conflict that did not happen — the client degrades that to UNKNOWN.
+			// The gRPC status code stays codes.Aborted because the transaction genuinely did abort;
+			// switching to Canceled or DeadlineExceeded would change retry behaviour for clients
+			// that already treat Aborted as retryable.
+			lateDetail = err.Error()
+		}
+		aborted = true
 	}
 	// Propose txn should be used to set watermark as done.
-	return s.proposeTxn(ctx, src)
+	if err := s.proposeTxn(ctx, src); err != nil {
+		return err
+	}
+	if aborted {
+		return status.Error(codes.Aborted, abortReason(lateReason, lateDetail))
+	}
+	// proposeTxn sets src.Aborted when it finds no commit timestamp for this transaction, meaning
+	// something aborted it out of band while this commit was in flight — a TryAbort from a schema
+	// update, a drop-predicate, or the idle-transaction reaper. Nothing above set `aborted`, because
+	// this commit itself decided to proceed. Returning nil here (as this did previously) loses the
+	// abort entirely: CommitOverNetwork then sees tctx.Aborted with no error and falls through to a
+	// bare dgo.ErrAborted, so the caller learns nothing. Report it instead.
+	if src.Aborted {
+		span.SetAttributes(attribute.Bool("abort", true))
+		return status.Error(codes.Aborted,
+			abortReason(abortReasonUncategorized, abortDetailPreAborted))
+	}
+	return nil
 }
 
 // CommitOrAbort either commits a transaction or aborts it.
diff --git a/dgraph/cmd/zero/oracle_reason_test.go b/dgraph/cmd/zero/oracle_reason_test.go
new file mode 100644
index 00000000000..1ad057b8774
--- /dev/null
+++ b/dgraph/cmd/zero/oracle_reason_test.go
@@ -0,0 +1,286 @@
+/*
+ * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package zero
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/dgraph-io/dgo/v250/protos/api"
+	"github.com/dgraph-io/dgraph/v25/protos/pb"
+)
+
+// The abort-reason wire format is a contract with gRPC clients (e.g. dgraph4j parses the
+// "<code>: " prefix into TxnConflictException.AbortReason). These unit tests pin the
+// category prefixes and the logic that selects between them, so the contract can't drift
+// silently without an integration cluster.
+
+func TestAbortReasonFormat(t *testing.T) {
+	require.Equal(t, "conflict: boom", abortReason(abortReasonConflict, "boom"))
+	require.Equal(t, "stale-startts: x", abortReason(abortReasonStaleStartTs, "x"))
+	require.Equal(t, "predicate-move: y", abortReason(abortReasonPredicateMove, "y"))
+}
+
+// TestAbortVocabulary pins the entire published surface: every category, and every detail a client
+// can receive for a server-decided abort, with the category each detail pairs with. Because the
+// details are declared in one block in oracle.go, adding a new abort message without deciding its
+// category — or quietly rewording an existing one that dgraph4j, pydgraph or the docs depend on —
+// fails here. The wording assertions are the pre-abort-reason text verbatim, so hoisting these to
+// constants is provably a move and not an edit.
+func TestAbortVocabulary(t *testing.T) {
+	// The three published codes, plus the withheld sentinel. Clients switch on exactly these.
+	require.Equal(t, "", abortReasonUncategorized)
+	require.Equal(t, "conflict", abortReasonConflict)
+	require.Equal(t, "stale-startts", abortReasonStaleStartTs)
+	require.Equal(t, "predicate-move", abortReasonPredicateMove)
+
+	for _, tc := range []struct {
+		name     string
+		category string
+		rendered string
+	}{
+		{"write-write conflict", abortReasonConflict,
+			abortDetailConflict},
+		{"stale start timestamp", abortReasonStaleStartTs,
+			abortDetailStaleStartTs},
+		{"move in flight", abortReasonPredicateMove,
+			fmt.Sprintf(abortDetailPredicateBlockedFmt, "friend")},
+		{"move completed", abortReasonPredicateMove,
+			fmt.Sprintf(abortDetailGroupMismatchFmt, 1, "friend", 2)},
+		{"predicate served by no group", abortReasonUncategorized,
+			fmt.Sprintf(abortDetailTabletNilFmt, "friend")},
+		{"malformed key, no separator", abortReasonUncategorized,
+			fmt.Sprintf(abortDetailMissingGroupIDFmt, "1foo")},
+		{"malformed key, bad group id", abortReasonUncategorized,
+			fmt.Sprintf(abortDetailBadGroupIDFmt, "xfoo")},
+		{"aborted out of band before commit", abortReasonUncategorized,
+			abortDetailPreAborted},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			wire := abortReason(tc.category, tc.rendered)
+			if tc.category == abortReasonUncategorized {
+				require.Equal(t, tc.rendered, wire, "a withheld cause must carry no prefix")
+				return
+			}
+			require.Equal(t, tc.category+": "+tc.rendered, wire)
+		})
+	}
+
+	// The conflict detail names what it cannot narrow — a conflict key is a one-way fingerprint by
+	// the time Zero matches it, so data, index, count and @upsert conflicts are indistinguishable.
+	// Callers who set one scalar value do not expect the keys derived from it to conflict, so the
+	// possibilities are spelled out.
+	for _, want := range []string{"index", "count", "@upsert"} {
+		require.Contains(t, abortDetailConflict, want,
+			"conflict detail should name the key kinds it cannot distinguish between")
+	}
+
+	// The out-of-band abort names its three causes rather than guessing a category. None of them is
+	// a predicate move — they all arrive via TryAbort, which records no cause — so labelling this
+	// predicate-move would be wrong. Before it was handled, commit() returned nil and the abort
+	// reached the client as a bare dgo.ErrAborted with no reason at all.
+	for _, want := range []string{"schema update", "drop-predicate", "txn-abort-after"} {
+		require.Contains(t, abortDetailPreAborted, want,
+			"pre-abort detail should name the causes it cannot distinguish between")
+	}
+	require.True(t, strings.HasPrefix(abortDetailPreAborted, "Transaction has been aborted. Please retry"),
+		"pre-abort detail must keep the legacy sentence as its prefix; got %q", abortDetailPreAborted)
+
+	// Backwards-compatibility guard. This exact sentence must remain a prefix of the conflict
+	// detail: dgraph/cmd/alpha/upsert_test.go loops while the live server error contains it (so a
+	// change would exit the loop immediately and fail), query/query4_test.go and
+	// systest/unique_test.go assert Contains on it, and it is verbatim the text of dgo.ErrAborted,
+	// which user log-scrapers key off. Extend abortDetailConflict by appending only.
+	require.True(t, strings.HasPrefix(abortDetailConflict, "Transaction has been aborted. Please retry"),
+		"conflict detail must keep the legacy sentence as its prefix; got %q", abortDetailConflict)
+
+	// Wording of the checkPreds details, verbatim from before the abort-reason work.
+	require.Equal(t, "Unable to find group id in 1foo",
+		fmt.Sprintf(abortDetailMissingGroupIDFmt, "1foo"))
+	require.Equal(t, "unable to parse group id from xfoo",
+		fmt.Sprintf(abortDetailBadGroupIDFmt, "xfoo"))
+	require.Equal(t, "Tablet for friend is nil",
+		fmt.Sprintf(abortDetailTabletNilFmt, "friend"))
+	require.Equal(t, "Commits on predicate friend are blocked due to predicate move",
+		fmt.Sprintf(abortDetailPredicateBlockedFmt, "friend"))
+	require.Equal(t, "Mutation done in group: 1. Predicate friend assigned to 2",
+		fmt.Sprintf(abortDetailGroupMismatchFmt, 1, "friend", 2))
+}
+
+// TestAbortReasonUncategorized pins the withholding contract: when the server cannot substantiate
+// one of the published categories it emits the bare detail with no prefix, rather than borrowing a
+// category that would imply the wrong remedy. That is byte-identical to what a pre-feature server
+// sends, and both dgraph4j and pydgraph already degrade a prefix-less message to UNKNOWN. The
+// absence of a colon-delimited prefix is the whole contract, so it must not regress.
+func TestAbortReasonUncategorized(t *testing.T) {
+	require.Equal(t, "Tablet for foo is nil",
+		abortReason(abortReasonUncategorized, "Tablet for foo is nil"))
+	require.Equal(t, "context canceled", abortReason(abortReasonUncategorized, "context canceled"))
+
+	// No withheld detail may begin with a token a client would parse as a real category, otherwise
+	// withholding would be silently reinterpreted as a category.
+	for _, detail := range []string{
+		"Unable to find group id in 1foo",
+		"Tablet for foo is nil",
+		"context canceled",
+		"context deadline exceeded",
+	} {
+		got := abortReason(abortReasonUncategorized, detail)
+		prefix := got
+		if i := strings.Index(got, ":"); i >= 0 {
+			prefix = got[:i]
+		}
+		for _, code := range []string{
+			abortReasonConflict, abortReasonStaleStartTs, abortReasonPredicateMove,
+		} {
+			require.NotEqual(t, code, strings.ToLower(strings.TrimSpace(prefix)),
+				"withheld detail %q would be parsed as category %q", detail, code)
+		}
+	}
+}
+
+func TestConflictAbortReason(t *testing.T) {
+	// Write-write conflict.
+	r := conflictAbortReason(false)
+	require.True(t, strings.HasPrefix(r, abortReasonConflict+": "),
+		"want conflict prefix, got %q", r)
+	require.Equal(t, abortReason(abortReasonConflict, abortDetailConflict), r)
+
+	// Stale start timestamp. The detail must name both ways startTxnTs rises — a Zero leader change
+	// and a conflict-map trim at a snapshot — because asserting only the first is wrong on the
+	// second path (purgeBelow via applySnapshot, which involves no leader change at all).
+	r = conflictAbortReason(true)
+	require.True(t, strings.HasPrefix(r, abortReasonStaleStartTs+": "),
+		"want stale-startts prefix, got %q", r)
+	require.Equal(t, abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs), r)
+	require.Contains(t, r, "leader change")
+	require.Contains(t, r, "snapshot")
+}
+
+// TestCheckPredsCategories is the core of the abort-category fix. checkPreds has five exits and only
+// two of them are predicate moves; the other three are unrelated failures with different remedies,
+// so they must not be reported as moves. Every case still returns an error (the transaction aborts
+// regardless) — this pins only which category each cause claims.
+func TestCheckPredsCategories(t *testing.T) {
+	const pred = "friend"
+	servingIn := func(gid uint32) *Server {
+		s := &Server{}
+		s.blockCommitsOn = new(sync.Map)
+		s.state = &pb.MembershipState{Groups: map[uint32]*pb.Group{
+			gid: {Tablets: map[string]*pb.Tablet{pred: {GroupId: gid, Predicate: pred}}},
+		}}
+		return s
+	}
+
+	t.Run("in-flight move is reported as a move", func(t *testing.T) {
+		// The authoritative signal, and it must win even though the tablet still resolves cleanly.
+		s := servingIn(1)
+		s.blockCommitsOn.Store(pred, struct{}{})
+		reason, err := s.checkPreds([]string{"1-" + pred})
+		require.Error(t, err)
+		require.Equal(t, abortReasonPredicateMove, reason)
+		require.Contains(t, err.Error(), "blocked due to predicate move")
+	})
+
+	t.Run("in-flight move wins over an absent tablet", func(t *testing.T) {
+		// Regression guard for the check ordering. blockTablet is held for the whole move, so if the
+		// tablet lookup ran first a genuine in-flight move could be reported as uncategorized.
+		s := &Server{state: &pb.MembershipState{Groups: map[uint32]*pb.Group{}}}
+		s.blockCommitsOn = new(sync.Map)
+		s.blockCommitsOn.Store(pred, struct{}{})
+		reason, err := s.checkPreds([]string{"1-" + pred})
+		require.Error(t, err)
+		require.Equal(t, abortReasonPredicateMove, reason,
+			"isBlocked must be consulted before the tablet lookup")
+	})
+
+	t.Run("completed move is reported as a move", func(t *testing.T) {
+		// Written against group 1, but the predicate now belongs to group 2.
+		reason, err := servingIn(2).checkPreds([]string{"1-" + pred})
+		require.Error(t, err)
+		require.Equal(t, abortReasonPredicateMove, reason)
+		require.Contains(t, err.Error(), "assigned to 2")
+	})
+
+	t.Run("predicate served by no group is not a move", func(t *testing.T) {
+		s := &Server{state: &pb.MembershipState{Groups: map[uint32]*pb.Group{}}}
+		s.blockCommitsOn = new(sync.Map)
+		reason, err := s.checkPreds([]string{"1-" + pred})
+		require.Error(t, err)
+		require.Equal(t, abortReasonUncategorized, reason,
+			"an unserved predicate is not a move; retrying may never succeed")
+	})
+
+	t.Run("malformed predicate key is not a move", func(t *testing.T) {
+		s := servingIn(1)
+		for _, pkey := range []string{pred, "x-" + pred} {
+			reason, err := s.checkPreds([]string{pkey})
+			require.Error(t, err, "pkey %q must abort", pkey)
+			require.Equal(t, abortReasonUncategorized, reason,
+				"a malformed predicate key %q is not a move and can never succeed on retry", pkey)
+		}
+	})
+
+	t.Run("healthy predicate does not abort", func(t *testing.T) {
+		reason, err := servingIn(1).checkPreds([]string{"1-" + pred})
+		require.NoError(t, err)
+		require.Equal(t, abortReasonUncategorized, reason)
+	})
+}
+
+// TestHasConflictStaleStartTs pins the exact discriminator commit() uses to choose the
+// stale-startts reason: a transaction whose startTs is below the leader's startTxnTs floor is a
+// conflict, and is flagged stale; a fresh startTs with no conflicting keys is neither.
+func TestHasConflictStaleStartTs(t *testing.T) {
+	o := &Oracle{}
+	o.Init()
+	defer o.close()
+
+	o.updateStartTxnTs(100)
+
+	// startTs below the floor: hasConflict true, and the stale discriminator true.
+	stale := &api.TxnContext{StartTs: 42}
+	require.True(t, o.hasConflict(stale), "transaction below startTxnTs must conflict")
+	require.True(t, o.isStaleStartTs(stale), "must be flagged stale")
+	require.Equal(t, conflictAbortReason(true), conflictAbortReason(o.isStaleStartTs(stale)))
+
+	// startTs at/above the floor with no keys: not a conflict, not stale.
+	fresh := &api.TxnContext{StartTs: 100}
+	require.False(t, o.hasConflict(fresh), "fresh transaction with no keys must not conflict")
+	require.False(t, o.isStaleStartTs(fresh), "must not be flagged stale")
+}
+
+// TestLateAbortStaleStartTs guards the late-abort path. Oracle.commit re-runs hasConflict and
+// collapses a stale start timestamp and a genuine keyCommit conflict into the same x.ErrConflict, so
+// before this fix every late abort was reported as "conflict" and the stale-startts category was
+// reachable only from the early check. Both causes are exercised here through the same discriminator
+// commit() now uses.
+func TestLateAbortStaleStartTs(t *testing.T) {
+	o := &Oracle{}
+	o.Init()
+	defer o.close()
+	o.updateStartTxnTs(100)
+
+	// Stale: Oracle.commit rejects it, and the late path must report stale-startts, not conflict.
+	stale := &api.TxnContext{StartTs: 42, CommitTs: 200}
+	require.Error(t, o.commit(stale), "a stale start timestamp must be rejected by Oracle.commit")
+	require.True(t, o.isStaleStartTs(stale))
+	require.True(t, strings.HasPrefix(conflictAbortReason(o.isStaleStartTs(stale)),
+		abortReasonStaleStartTs+": "), "late stale abort must not be labelled a conflict")
+
+	// A genuine write-write conflict above the floor still reports conflict.
+	first := &api.TxnContext{StartTs: 100, CommitTs: 150, Keys: []string{"a"}}
+	require.NoError(t, o.commit(first))
+	second := &api.TxnContext{StartTs: 120, CommitTs: 200, Keys: []string{"a"}}
+	require.Error(t, o.commit(second), "second writer of key a must conflict")
+	require.False(t, o.isStaleStartTs(second), "above the floor, so not stale")
+	require.True(t, strings.HasPrefix(conflictAbortReason(o.isStaleStartTs(second)),
+		abortReasonConflict+": "), "a real write-write conflict must stay a conflict")
+}
diff --git a/edgraph/server.go b/edgraph/server.go
index e5e9de7164c..69083f585f2 100644
--- a/edgraph/server.go
+++ b/edgraph/server.go
@@ -700,6 +700,10 @@ func (s *Server) doMutate(ctx context.Context, qc *queryContext, resp *api.Respo
 		if err == dgo.ErrAborted {
 			err = status.Error(codes.Aborted, err.Error())
 			resp.Txn.Aborted = true
+		} else if status.Code(err) == codes.Aborted {
+			// Server-decided abort carrying a categorized reason; err is already a codes.Aborted
+			// status (e.g. "conflict: ...", "predicate-move: ...") — surface it unchanged.
+			resp.Txn.Aborted = true
 		}
 
 		return err
@@ -2139,6 +2143,12 @@ func (s *Server) CommitOrAbort(ctx context.Context, tc *api.TxnContext) (*api.Tx
 
 		return tctx, status.Error(codes.Aborted, err.Error())
 	}
+	if status.Code(err) == codes.Aborted {
+		// Server-decided abort carrying a categorized reason; err is already a codes.Aborted
+		// status (e.g. "conflict: ...", "predicate-move: ...") — surface it unchanged.
+		tctx.Aborted = true
+		return tctx, err
+	}
 	tctx.StartTs = tc.StartTs
 	tctx.CommitTs = commitTs
 	return tctx, err
diff --git a/systest/integration2/txn_abort_reason_test.go b/systest/integration2/txn_abort_reason_test.go
new file mode 100644
index 00000000000..342577c444c
--- /dev/null
+++ b/systest/integration2/txn_abort_reason_test.go
@@ -0,0 +1,88 @@
+//go:build integration2
+
+/*
+ * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+
+	"github.com/dgraph-io/dgo/v250/protos/api"
+	"github.com/dgraph-io/dgraph/v25/dgraphtest"
+)
+
+// TestStaleStartTsAbortReason proves the "stale-startts" abort category end-to-end against a
+// real cluster. A transaction's start timestamp becomes "stale" when it predates the current
+// Zero leader's lease — i.e. after a leader change. We reproduce that deterministically by
+// opening a transaction, then restarting the (single) Zero: on restart Zero renews its lease and
+// advances startTxnTs past every previously-leased start timestamp, so committing the now-old
+// transaction aborts
+// with the stale-startts reason rather than a plain write-write conflict.
+//
+// As in the alpha-level reason tests, the commit goes through the raw CommitOrAbort stub so we
+// observe the unflattened codes.Aborted status (dgo's Txn.Commit would replace it with the
+// reasonless ErrAborted).
+func TestStaleStartTsAbortReason(t *testing.T) {
+	conf := dgraphtest.NewClusterConfig().WithNumAlphas(1).WithNumZeros(1).WithReplicas(1)
+	c, err := dgraphtest.NewLocalCluster(conf)
+	require.NoError(t, err)
+	t.Cleanup(func() { c.Cleanup(t.Failed()) })
+	require.NoError(t, c.Start())
+
+	gc, cleanup, err := c.Client()
+	require.NoError(t, err)
+	defer cleanup()
+
+	ctx := context.Background()
+	require.NoError(t, gc.Alter(ctx, &api.Operation{DropAll: true}))
+
+	// Open a transaction and mutate so it gets a real (soon-to-be-stale) start timestamp and keys.
+	txn := gc.NewTxn()
+	resp, err := txn.Mutate(ctx, &api.Mutation{SetJson: []byte(`{"name": "Manish"}`)})
+	require.NoError(t, err)
+	tc := resp.GetTxn()
+	require.NotZero(t, tc.GetStartTs(), "mutation must yield a start ts")
+
+	// Restart Zero. On coming back up it renews its lease and sets startTxnTs to MaxTxnTs+1,
+	// which is strictly greater than the start timestamp leased above — making our open
+	// transaction stale.
+	require.NoError(t, c.StopZero(0))
+	require.NoError(t, c.StartZero(0))
+	require.NoError(t, c.HealthCheck(false))
+
+	// Wait until a Zero leader is established again (lease renewal, hence startTxnTs bump, runs
+	// when a Zero becomes leader). Avoids racing the commit against the leaderless window.
+	require.Eventually(t, func() bool {
+		_, err := c.GetZeroLeader(0)
+		return err == nil
+	}, 60*time.Second, time.Second, "zero leader did not re-establish after restart")
+
+	// Commit the stale transaction via the raw stub to observe the categorized status.
+	port, err := c.GetAlphaGrpcPublicPort(0)
+	require.NoError(t, err)
+	conn, err := grpc.NewClient("0.0.0.0:"+port, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() { _ = conn.Close() }()
+	raw := api.NewDgraphClient(conn)
+
+	_, err = raw.CommitOrAbort(ctx, tc)
+	require.Error(t, err, "committing a txn whose start ts predates the new leader must abort")
+
+	st := status.Convert(err)
+	require.Equal(t, codes.Aborted, st.Code(),
+		"abort must keep codes.Aborted so existing clients still retry")
+	require.True(t, strings.HasPrefix(st.Message(), "stale-startts: "),
+		"abort reason should be categorized as stale-startts; got %q", st.Message())
+}
diff --git a/worker/mutation.go b/worker/mutation.go
index 5efc6b2d1d7..b1babce4c84 100644
--- a/worker/mutation.go
+++ b/worker/mutation.go
@@ -20,6 +20,8 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/dgraph-io/badger/v4"
@@ -854,6 +856,16 @@ func CommitOverNetwork(ctx context.Context, tc *api.TxnContext) (uint64, error)
 	h := hooks.GetHooks()
 	tctx, err := h.CommitOrAbort(ctx, tc)
 	if err != nil {
+		// Zero signals a server-decided abort as a codes.Aborted status carrying a categorized
+		// reason (e.g. "conflict: ...", "predicate-move: ..."). Record the abort metric and
+		// forward the reason rather than treating it as a generic transport error or flattening
+		// it to the reasonless dgo.ErrAborted.
+		if status.Code(err) == codes.Aborted {
+			if !clientDiscard {
+				ostats.Record(ctx, x.TxnAborts.M(1))
+			}
+			return 0, err
+		}
 		span.AddEvent("Error in CommitOrAbort", trace.WithAttributes(
 			attribute.String("error", err.Error())))
 		return 0, err