Serialize account credential and session changes

Why: Concurrent password changes, recovery key reissue, and session admission could accept retired credentials or omit rejected audits.

Evidence: Local Go vet and full tests passed; real PostgreSQL store suite, deterministic interleavings, and two-process contention passed with SELF receipts.

Impact: Account writes and session admission serialize per owner; stale sessions fail closed; checked rejections carry an audit row.

Refs: Agent-System-Dev #486; Agent-System-Backend #185; merge queue Issue #518

internal/app/app.go

@@ -959,7 +959,7 @@ func VerifySystemIdentity(storageDSN, dataDir, instanceID, dbref string) (int, e
 }
 
 func cliActor() string {
-	return fmt.Sprintf("uid:%d", os.Getuid())
+	return fmt.Sprintf("%d", os.Getuid())
 }
 
 func accountInstanceID() string {

internal/store/account_process_crash_test.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
@@ -27,9 +28,8 @@ func TestAccountPasswordSurvivesProcessTermination(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		if err := st.Close(); err != nil {
-			t.Fatal(err)
-		}
+		// Keep the database connection and process alive at the signal point;
+		// a graceful close would only test reopen, not abrupt termination.
 		fmt.Println("U7_COMMITTED")
 		select {}
 	}
@@ -82,7 +82,15 @@ func TestAccountPasswordSurvivesProcessTermination(t *testing.T) {
 			if err := child.Process.Signal(tc.signal); err != nil {
 				t.Fatal(err)
 			}
-			_ = child.Wait()
+			waitErr := child.Wait()
+			var exitErr *exec.ExitError
+			if !errors.As(waitErr, &exitErr) {
+				t.Fatalf("child was not terminated by %s: %v", tc.name, waitErr)
+			}
+			status, ok := exitErr.Sys().(syscall.WaitStatus)
+			if !ok || !status.Signaled() || status.Signal() != tc.signal {
+				t.Fatalf("child ended without requested %s: %v", tc.name, waitErr)
+			}
 			reopened, err := OpenExistingSQLite(path)
 			if err != nil {
 				t.Fatal(err)

internal/store/account_security_integration_test.go

@@ -101,3 +101,87 @@ func TestRevokeAllRefusesMissingOwnerWithoutCredential(t *testing.T) {
 		t.Fatalf("orphan credential rows=%d err=%v", n, err)
 	}
 }
+
+func TestActivePasswordChangeRevokesUnusedRecoveryKey(t *testing.T) {
+	s := activeAccount(t)
+	key, _, err := s.IssueLoginKey("ada", LoginKeyPurposeRecover, "admin", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	session, err := s.CreateSession("ada")
+	if err != nil {
+		t.Fatal(err)
+	}
+	hash, err := HashPassword("replacement-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ChangeUserPasswordActive("ada", "old-password", hash, session); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := s.RedeemLoginKey(key); err == nil {
+		t.Fatal("unused recovery key issued before active password change still redeems")
+	}
+}
+
+func TestRejectedActiveChangeIsAudited(t *testing.T) {
+	s := activeAccount(t)
+	h, err := HashPassword("next-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = s.ChangeUserPasswordActive("ada", "incorrect-current-password", h, "")
+	if !errors.Is(err, ErrBadCredentials) {
+		t.Fatalf("expected bad credentials, got %v", err)
+	}
+	count, err := s.CountAudit("ada", "rejected")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 {
+		t.Fatalf("rejected checked account mutation has %d rejected audit rows, want1", count)
+	}
+}
+
+func TestDelayedOldVersionSessionIsRejectedAtUse(t *testing.T) {
+	s := activeAccount(t)
+	_, ver, ok := s.VerifyLoginWithVersion("ada", "old-password")
+	if !ok {
+		t.Fatal("healthy verification missing")
+	}
+	h, err := HashPassword("new-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.ApplyPasswordChange(AccountWrite{Owner: "ada", PasswordHash: h}); err != nil {
+		t.Fatal(err)
+	}
+	// Represents an insert racing after the replacement's earlier DELETE.
+	token := "sess_private_stale_fixture"
+	_, err = s.db.Exec(s.q(`INSERT INTO sessions(token_hash,owner,created_at,expires_at,last_seen_at,scope,auth_version) VALUES(?,?,?,?,?,?,?)`),
+		hashKey(token), "ada", nowISO(), time.Now().Add(time.Hour).UTC().Format(time.RFC3339Nano), nowISO(), SessionScopeFull, ver)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, _, ok := s.SessionInfo(token); ok {
+		t.Fatal("session carrying the retired auth_version became authorized")
+	}
+}
+
+func TestActivePasswordChangePreservesCurrentVersionedSession(t *testing.T) {
+	s := activeAccount(t)
+	token, err := s.CreateSession("ada")
+	if err != nil {
+		t.Fatal(err)
+	}
+	h, err := HashPassword("new-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ChangeUserPasswordActive("ada", "old-password", h, token); err != nil {
+		t.Fatal(err)
+	}
+	if owner, scope, ok := s.SessionInfo(token); !ok || owner != "ada" || scope != SessionScopeFull {
+		t.Fatal("active change did not retain its current full session")
+	}
+}

internal/store/account_serialization_pg_test.go

@@ -0,0 +1,167 @@
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// The trigger holds a transaction after its authorization read. A second
+// connection must wait for that owner's lock before replacing credentials.
+func holdPGAccountWrite(t *testing.T, s *Store, table, event, suffix string) func() {
+	t.Helper()
+	lock := "u7_account_" + suffix
+	_, err := s.db.Exec(`CREATE FUNCTION u7_pause_account() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_advisory_xact_lock(hashtext(current_schema() || ':` + lock + `')); RETURN NEW; END $$`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = s.db.Exec(`CREATE TRIGGER u7_pause_account BEFORE ` + event + ` ON ` + table + ` FOR EACH ROW EXECUTE FUNCTION u7_pause_account()`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn, err := s.db.Conn(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := conn.ExecContext(context.Background(), `SELECT pg_advisory_lock(hashtext(current_schema() || ':`+lock+`'))`); err != nil {
+		conn.Close()
+		t.Fatal(err)
+	}
+	released := false
+	release := func() {
+		if !released {
+			released = true
+			_, _ = conn.ExecContext(context.Background(), `SELECT pg_advisory_unlock(hashtext(current_schema() || ':`+lock+`'))`)
+			_ = conn.Close()
+		}
+	}
+	t.Cleanup(release)
+	return release
+}
+
+func waitForPGAccountWrite(t *testing.T, s *Store, suffix string) {
+	t.Helper()
+	lock := "u7_account_" + suffix
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		var n int
+		err := s.db.QueryRow(`SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted
+			AND objid=(hashtext(current_schema() || ':` + lock + `')::bigint & 4294967295)::oid`).Scan(&n)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if n > 0 {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("account write did not reach the deterministic PostgreSQL pause")
+}
+
+func awaitPGAccountResult[T any](t *testing.T, done <-chan T) T {
+	t.Helper()
+	select {
+	case result := <-done:
+		return result
+	case <-time.After(5 * time.Second):
+		t.Fatal("serialized account transaction did not finish")
+		var zero T
+		return zero
+	}
+}
+
+func TestPGVerifiedSessionAndPasswordChangeSerialize(t *testing.T) {
+	s := activeAccount(t)
+	if s.Dialect() != "postgres" {
+		t.Skip("real PostgreSQL required")
+	}
+	_, version, ok := s.VerifyLoginWithVersion("ada", "old-password")
+	if !ok {
+		t.Fatal("initial login failed")
+	}
+	release := holdPGAccountWrite(t, s, "sessions", "INSERT", "session")
+	type sessionResult struct {
+		token string
+		err   error
+	}
+	sessionDone := make(chan sessionResult, 1)
+	go func() {
+		token, err := s.CreateSessionVerified("ada", SessionScopeFull, time.Hour, version)
+		sessionDone <- sessionResult{token, err}
+	}()
+	waitForPGAccountWrite(t, s, "session")
+	hash, err := HashPassword("replacement-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	changeDone := make(chan error, 1)
+	go func() {
+		_, err := s.ApplyPasswordChange(AccountWrite{Owner: "ada", PasswordHash: hash, Actor: "admin"})
+		changeDone <- err
+	}()
+	select {
+	case err := <-changeDone:
+		t.Fatalf("replacement committed ahead of the in-flight verified session: %v", err)
+	case <-time.After(100 * time.Millisecond):
+	}
+	release()
+	session := awaitPGAccountResult(t, sessionDone)
+	if session.err != nil {
+		t.Fatal(session.err)
+	}
+	if err := awaitPGAccountResult(t, changeDone); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, ok := s.SessionInfo(session.token); ok {
+		t.Fatal("old verified session survived later password replacement")
+	}
+}
+
+func TestPGRestrictedPasswordSetAndKeyReissueSerialize(t *testing.T) {
+	s := openTestStore(t)
+	if s.Dialect() != "postgres" {
+		t.Skip("real PostgreSQL required")
+	}
+	if err := s.CreateAccount("ada", "Ada", "", false); err != nil {
+		t.Fatal(err)
+	}
+	key, _, err := s.IssueLoginKey("ada", LoginKeyPurposeRecover, "admin", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	token, _, err := s.RedeemLoginKey(key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	hash, err := HashPassword("replacement-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	release := holdPGAccountWrite(t, s, "users", "UPDATE", "password")
+	setDone := make(chan error, 1)
+	go func() {
+		_, err := s.CompleteSetPasswordChecked(token, hash)
+		setDone <- err
+	}()
+	waitForPGAccountWrite(t, s, "password")
+	issueDone := make(chan error, 1)
+	go func() {
+		_, _, err := s.IssueLoginKey("ada", LoginKeyPurposeRecover, "admin", 0)
+		issueDone <- err
+	}()
+	select {
+	case err := <-issueDone:
+		t.Fatalf("key replacement committed ahead of the in-flight password set: %v", err)
+	case <-time.After(100 * time.Millisecond):
+	}
+	release()
+	if err := awaitPGAccountResult(t, setDone); err != nil {
+		t.Fatal(err)
+	}
+	if err := awaitPGAccountResult(t, issueDone); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, ok := s.SessionInfo(token); ok {
+		t.Fatal("old restricted session survived password set and key reissue")
+	}
+}

internal/store/account_two_process_pg_test.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"bytes"
+	"context"
 	"database/sql"
 	"errors"
 	"fmt"
@@ -24,6 +25,9 @@ func TestAccountTwoProcessPostgresAdmission(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer st.Close()
+		if err := os.WriteFile(os.Getenv("U7_ACCOUNT_CHILD_READY"), []byte("ready"), 0o600); err != nil {
+			t.Fatal(err)
+		}
 		for i := 0; i < 500; i++ {
 			if _, err := os.Stat(barrier); err == nil {
 				break
@@ -96,14 +100,38 @@ func TestAccountTwoProcessPostgresAdmission(t *testing.T) {
 				t.Fatal(err)
 			}
 			st.Close()
+			// Hold the shared admission point so both independent processes
+			// are demonstrably contending before either may commit.
+			lockKey := "bootstrap:first-admin"
+			if tc.mode == "create" {
+				lockKey = "account:same"
+			}
+			lockConn, err := admin.Conn(context.Background())
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer lockConn.Close()
+			if _, err := lockConn.ExecContext(context.Background(), `SELECT pg_advisory_lock(hashtext($1))`, lockKey); err != nil {
+				t.Fatal(err)
+			}
+			released := false
+			release := func() {
+				if !released {
+					released = true
+					_, _ = lockConn.ExecContext(context.Background(), `SELECT pg_advisory_unlock(hashtext($1))`, lockKey)
+				}
+			}
+			defer release()
 			barrier := filepath.Join(t.TempDir(), "go")
+			readyDir := t.TempDir()
 			var children [2]*exec.Cmd
 			var logs [2]bytes.Buffer
 			for i := range children {
+				readyPath := filepath.Join(readyDir, fmt.Sprintf("ready-%d", i))
 				cmd := exec.Command(os.Args[0], "-test.run=^TestAccountTwoProcessPostgresAdmission$", "-test.v")
 				cmd.Env = append(os.Environ(), "U7_ACCOUNT_CHILD_MODE="+tc.mode, "U7_ACCOUNT_CHILD_DSN="+dsn,
 					"U7_ACCOUNT_CHILD_OWNER="+tc.owners[i], "U7_ACCOUNT_CHILD_ARTIFACTS="+artifacts,
-					"U7_ACCOUNT_CHILD_BARRIER="+barrier)
+					"U7_ACCOUNT_CHILD_BARRIER="+barrier, "U7_ACCOUNT_CHILD_READY="+readyPath)
 				cmd.Stdout = &logs[i]
 				cmd.Stderr = &logs[i]
 				if err := cmd.Start(); err != nil {
@@ -111,9 +139,38 @@ func TestAccountTwoProcessPostgresAdmission(t *testing.T) {
 				}
 				children[i] = cmd
 			}
+			deadline := time.Now().Add(10 * time.Second)
+			for i := range children {
+				readyPath := filepath.Join(readyDir, fmt.Sprintf("ready-%d", i))
+				for {
+					if _, err := os.Stat(readyPath); err == nil {
+						break
+					}
+					if time.Now().After(deadline) {
+						t.Fatalf("child %d never opened the shared database: %s", i, logs[i].String())
+					}
+					time.Sleep(5 * time.Millisecond)
+				}
+			}
 			if err := os.WriteFile(barrier, []byte("go"), 0o600); err != nil {
 				t.Fatal(err)
 			}
+			for {
+				var waiters int
+				err := admin.QueryRow(`SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted
+					AND objid=(hashtext($1)::bigint & 4294967295)::oid`, lockKey).Scan(&waiters)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if waiters >= 2 {
+					break
+				}
+				if time.Now().After(deadline) {
+					t.Fatalf("only %d of 2 child processes reached the shared admission lock: %s | %s", waiters, logs[0].String(), logs[1].String())
+				}
+				time.Sleep(5 * time.Millisecond)
+			}
+			release()
 			wins, losses := 0, 0
 			for i, cmd := range children {
 				if err := cmd.Wait(); err != nil {

internal/store/account_tx.go

@@ -104,6 +104,12 @@ func (s *Store) ApplyPasswordChange(w AccountWrite) (AccountOutcome, error) {
 		if err != nil {
 			return err
 		}
+		if w.ExceptToken != "" {
+			if _, err := tx.Exec(s.q(`UPDATE sessions SET auth_version=(SELECT auth_version FROM users WHERE owner=?) WHERE owner=? AND token_hash=?`),
+				w.Owner, w.Owner, hashKey(w.ExceptToken)); err != nil {
+				return err
+			}
+		}
 		out.OperationID = opID
 		return nil
 	})
@@ -204,6 +210,16 @@ func (s *Store) accountTx(w AccountWrite, fn func(*sql.Tx, string) error) error
 			_ = tx.Rollback()
 		}
 	}()
+	if w.Owner != "" {
+		if err := s.dialect.lockAccount(tx, w.Owner); err != nil {
+			return err
+		}
+	}
+	if w.InstanceID == "" {
+		if err := tx.QueryRow(`SELECT instance_id FROM agent_system_identity`).Scan(&w.InstanceID); err != nil && !errors.Is(err, sql.ErrNoRows) {
+			return err
+		}
+	}
 	if err := fn(tx, opID); err != nil {
 		_ = tx.Rollback()
 		committed = true
@@ -218,10 +234,14 @@ func (s *Store) accountTx(w AccountWrite, fn func(*sql.Tx, string) error) error
 		return err
 	}
 	if err := insertAudit(s, tx, w, opID, "success"); err != nil {
+		_ = tx.Rollback()
+		committed = true
+		s.recordRejected(w, opID)
 		return err
 	}
 	if err := tx.Commit(); err != nil {
-		s.recordRejected(w, opID)
+		// Commit may have reached the database even if the client lost its
+		// response. Recording a rejection here would assert an unknown outcome.
 		return err
 	}
 	committed = true
@@ -350,31 +370,3 @@ func (s *Store) CountAudit(owner, outcome string) (int, error) {
 	err := s.db.QueryRow(s.q(`SELECT COUNT(1) FROM audit_events WHERE owner=? AND entry=? AND outcome=?`), owner, "account", outcome).Scan(&n)
 	return n, err
 }
-
-// commitAccountMutation extends existing checked transactions without moving
-// their authorization checks outside the transaction or replacing account_events.
-func (s *Store) commitAccountMutation(tx *sql.Tx, w AccountWrite) error {
-	opID := newID("op")
-	if w.InstanceID == "" {
-		err := tx.QueryRow(`SELECT instance_id FROM agent_system_identity`).Scan(&w.InstanceID)
-		if err != nil && !errors.Is(err, sql.ErrNoRows) {
-			return err
-		}
-	}
-	if s.accountFault != nil {
-		err := s.accountFault
-		_ = tx.Rollback()
-		s.recordRejected(w, opID)
-		return err
-	}
-	if err := insertAudit(s, tx, w, opID, "success"); err != nil {
-		_ = tx.Rollback()
-		s.recordRejected(w, opID)
-		return err
-	}
-	if err := tx.Commit(); err != nil {
-		s.recordRejected(w, opID)
-		return err
-	}
-	return nil
-}

internal/store/auth.go

@@ -313,17 +313,32 @@ func (s *Store) CreateSessionScoped(owner, scope string, ttl time.Duration) (str
 	if ttl <= 0 {
 		ttl = sessionTTL
 	}
-	ver := 0
-	_ = s.db.QueryRow(s.q(`SELECT auth_version FROM users WHERE owner=?`), owner).Scan(&ver)
 	raw := make([]byte, 32)
 	if _, err := rand.Read(raw); err != nil {
 		return "", err
 	}
 	token := "sess_" + hex.EncodeToString(raw)
 	exp := time.Now().Add(ttl)
-	_, err := s.db.Exec(s.q(`INSERT INTO sessions(token_hash,owner,created_at,expires_at,last_seen_at,scope,auth_version) VALUES(?,?,?,?,?,?,?)`),
+	tx, err := s.db.Begin()
+	if err != nil {
+		return "", err
+	}
+	defer tx.Rollback()
+	if err := s.dialect.lockAccount(tx, owner); err != nil {
+		return "", err
+	}
+	var ver int
+	if err := tx.QueryRow(s.q(`SELECT auth_version FROM users WHERE owner=?`), owner).Scan(&ver); errors.Is(err, sql.ErrNoRows) {
+		return "", ErrNotFound
+	} else if err != nil {
+		return "", err
+	}
+	_, err = tx.Exec(s.q(`INSERT INTO sessions(token_hash,owner,created_at,expires_at,last_seen_at,scope,auth_version) VALUES(?,?,?,?,?,?,?)`),
 		hashKey(token), owner, nowISO(), exp.UTC().Format(time.RFC3339Nano), nowISO(), scope, ver)
-	return token, err
+	if err != nil {
+		return "", err
+	}
+	return token, tx.Commit()
 }
 
 // CreateSessionVerified 建会话并绑定期望认证版本：事务内重读账户版本，与密码验证
@@ -346,6 +361,9 @@ func (s *Store) CreateSessionVerified(owner, scope string, ttl time.Duration, ex
 		return "", err
 	}
 	defer tx.Rollback()
+	if err := s.dialect.lockAccount(tx, owner); err != nil {
+		return "", err
+	}
 	var ver int
 	err = tx.QueryRow(s.q(`SELECT auth_version FROM users WHERE owner=?`), owner).Scan(&ver)
 	if errors.Is(err, sql.ErrNoRows) {
@@ -376,11 +394,17 @@ func (s *Store) SessionInfo(token string) (owner, scope string, ok bool) {
 		return "", "", false
 	}
 	var exp string
-	err := s.db.QueryRow(s.q(`SELECT owner,expires_at,COALESCE(scope,'full') FROM sessions WHERE token_hash=?`), hashKey(token)).
-		Scan(&owner, &exp, &scope)
+	var sessionVersion, accountVersion int
+	err := s.db.QueryRow(s.q(`SELECT s.owner,s.expires_at,COALESCE(s.scope,'full'),s.auth_version,u.auth_version
+		FROM sessions s JOIN users u ON u.owner=s.owner WHERE s.token_hash=?`), hashKey(token)).
+		Scan(&owner, &exp, &scope, &sessionVersion, &accountVersion)
 	if err != nil {
 		return "", "", false
 	}
+	if sessionVersion != accountVersion {
+		_, _ = s.db.Exec(s.q(`DELETE FROM sessions WHERE token_hash=? AND auth_version=?`), hashKey(token), sessionVersion)
+		return "", "", false
+	}
 	if expT, err := time.Parse(time.RFC3339Nano, exp); err != nil || time.Now().After(expT) {
 		s.db.Exec(s.q(`DELETE FROM sessions WHERE token_hash=?`), hashKey(token))
 		return "", "", false

internal/store/dialect.go

@@ -20,6 +20,8 @@ type dialect interface {
 	// lockBootstrap 锁定首次管理员引导的「核验无用户→建号→激活 Key」串行段
 	//（KTD12：防两个进程并发引导出双管理员）。
 	lockBootstrap(tx *sql.Tx) error
+	// lockAccount serializes credential and session admission for one owner.
+	lockAccount(tx *sql.Tx, owner string) error
 }
 
 // ---- SQLite（嵌入式档）----
@@ -32,6 +34,7 @@ func (sqliteDialect) lockConversation(*sql.Tx, string) error  { return nil }
 func (sqliteDialect) lockScope(*sql.Tx, string, string) error { return nil }
 func (sqliteDialect) lockAdmission(*sql.Tx) error             { return nil }
 func (sqliteDialect) lockBootstrap(*sql.Tx) error             { return nil }
+func (sqliteDialect) lockAccount(*sql.Tx, string) error       { return nil }
 
 // ---- PostgreSQL（多用户/多实例档）----
 
@@ -72,3 +75,8 @@ func (postgresDialect) lockBootstrap(tx *sql.Tx) error {
 	_, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext($1))`, "bootstrap:first-admin")
 	return err
 }
+
+func (postgresDialect) lockAccount(tx *sql.Tx, owner string) error {
+	_, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext($1))`, "account:"+owner)
+	return err
+}

internal/store/login_keys.go

@@ -112,6 +112,15 @@ func (s *Store) IssueLoginKey(owner, purpose, createdBy string, ttl time.Duratio
 		return "", meta, err
 	}
 	defer tx.Rollback()
+	if err := s.dialect.lockAccount(tx, owner); err != nil {
+		return "", meta, err
+	}
+	var exists int
+	if err := tx.QueryRow(s.q(`SELECT 1 FROM users WHERE owner=?`), owner).Scan(&exists); errors.Is(err, sql.ErrNoRows) {
+		return "", meta, ErrNotFound
+	} else if err != nil {
+		return "", meta, err
+	}
 	meta, err = s.insertLoginKeyTx(tx, owner, purpose, createdBy, plain, id, now, exp)
 	if err != nil {
 		return "", meta, err
@@ -204,11 +213,20 @@ func (s *Store) RevokeLoginKey(id string) error {
 // RevokeLoginKeyCascade 撤销 Key 并作废该用户全部 set_password_only 会话（管理员
 // HTTP 入口）。Key 已消费/过期时仍返回属主并作废会话；编号不存在才报 NotFound。
 func (s *Store) RevokeLoginKeyCascade(id, actor string) (owner string, err error) {
+	if err := s.db.QueryRow(s.q(`SELECT owner FROM login_keys WHERE id=?`), id).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
+		return "", ErrNotFound
+	} else if err != nil {
+		return "", err
+	}
 	tx, err := s.db.Begin()
 	if err != nil {
 		return "", err
 	}
 	defer tx.Rollback()
+	if err := s.dialect.lockAccount(tx, owner); err != nil {
+		return "", err
+	}
+	lockedOwner := owner
 	err = tx.QueryRow(s.q(`SELECT owner FROM login_keys WHERE id=?`), id).Scan(&owner)
 	if errors.Is(err, sql.ErrNoRows) {
 		return "", ErrNotFound
@@ -216,6 +234,9 @@ func (s *Store) RevokeLoginKeyCascade(id, actor string) (owner string, err error
 	if err != nil {
 		return "", err
 	}
+	if owner != lockedOwner {
+		return "", ErrConflict
+	}
 	if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE id=? AND status=?`),
 		LoginKeyStatusRevoked, id, LoginKeyStatusUnused); err != nil {
 		return "", err
@@ -279,11 +300,20 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er
 	if plain == "" || len(plain) < 5 || plain[:4] != "alk_" {
 		return "", "", ErrNotFound
 	}
+	var lockOwner string
+	if err := s.db.QueryRow(s.q(`SELECT owner FROM login_keys WHERE key_hash=?`), hashKey(plain)).Scan(&lockOwner); errors.Is(err, sql.ErrNoRows) {
+		return "", "", ErrNotFound
+	} else if err != nil {
+		return "", "", err
+	}
 	tx, err := s.db.Begin()
 	if err != nil {
 		return "", "", err
 	}
 	defer tx.Rollback()
+	if err := s.dialect.lockAccount(tx, lockOwner); err != nil {
+		return "", "", err
+	}
 	var id, o, status, exp string
 	err = tx.QueryRow(s.q(`SELECT id,owner,status,expires_at FROM login_keys WHERE key_hash=?`), hashKey(plain)).
 		Scan(&id, &o, &status, &exp)
@@ -293,6 +323,9 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er
 	if err != nil {
 		return "", "", err
 	}
+	if o != lockOwner {
+		return "", "", ErrConflict
+	}
 	expT, err := time.Parse(time.RFC3339Nano, exp)
 	if err != nil || time.Now().After(expT) {
 		_, _ = tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE id=? AND status=?`), LoginKeyStatusExpired, id, LoginKeyStatusUnused)
@@ -339,137 +372,139 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er
 // 写 hash、清 must_change、bump auth_version、作废未用登录 Key、删除全部会话、
 // 写成功审计。会话或版本失效时返回 ErrConflict，不做任何写入。
 func (s *Store) CompleteSetPasswordChecked(token, passwordHash string) (owner string, err error) {
-	tx, err := s.db.Begin()
+	// Discover only the lock key here. Authorization is checked again after
+	// acquiring the owner's transaction lock, alongside key reissue/revocation.
+	if err := s.db.QueryRow(s.q(`SELECT owner FROM sessions WHERE token_hash=?`), hashKey(token)).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
+		return "", ErrNotFound
+	} else if err != nil {
+		return "", err
+	}
+	lockOwner := owner
+	err = s.accountTx(AccountWrite{Owner: owner, Actor: owner, Action: "account.set_password"}, func(tx *sql.Tx, _ string) error {
+		var scope, sessExp, actualOwner string
+		var sessVer, userVer int
+		var mustChange bool
+		var hash sql.NullString
+		err := tx.QueryRow(s.q(`SELECT s.owner,s.scope,s.expires_at,s.auth_version,u.auth_version,u.must_change_password,u.password_hash
+			FROM sessions s JOIN users u ON u.owner=s.owner WHERE s.token_hash=?`), hashKey(token)).
+			Scan(&actualOwner, &scope, &sessExp, &sessVer, &userVer, &mustChange, &hash)
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		if err != nil {
+			return err
+		}
+		if actualOwner != lockOwner {
+			return ErrConflict
+		}
+		if expT, perr := time.Parse(time.RFC3339Nano, sessExp); perr != nil || time.Now().After(expT) {
+			return ErrNotFound
+		}
+		if scope != SessionScopeSetPasswordOnly && !mustChange && hash.Valid && hash.String != "" {
+			return ErrConflict
+		}
+		if sessVer != userVer {
+			return ErrConflict
+		}
+		res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`),
+			passwordHash, nowISO(), false, lockOwner)
+		if err != nil {
+			return err
+		}
+		if n, _ := res.RowsAffected(); n != 1 {
+			return ErrNotFound
+		}
+		if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`),
+			LoginKeyStatusRevoked, lockOwner, LoginKeyStatusUnused); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), lockOwner); err != nil {
+			return err
+		}
+		entrance := "must-change"
+		if scope == SessionScopeSetPasswordOnly {
+			entrance = "recovery"
+		}
+		return s.insertAccountEvent(tx, lockOwner, entrance, lockOwner, "password_set", entrance)
+	})
 	if err != nil {
 		return "", err
 	}
-	defer tx.Rollback()
-	var scope, sessExp string
-	var sessVer, userVer int
-	var mustChange bool
-	var hash sql.NullString
-	err = tx.QueryRow(s.q(`SELECT s.owner,s.scope,s.expires_at,s.auth_version,u.auth_version,u.must_change_password,u.password_hash
-		FROM sessions s JOIN users u ON u.owner=s.owner WHERE s.token_hash=?`), hashKey(token)).
-		Scan(&owner, &scope, &sessExp, &sessVer, &userVer, &mustChange, &hash)
-	if errors.Is(err, sql.ErrNoRows) {
-		return "", ErrNotFound
-	}
-	if err != nil {
-		return "", err
-	}
-	if expT, perr := time.Parse(time.RFC3339Nano, sessExp); perr != nil || time.Now().After(expT) {
-		return "", ErrNotFound
-	}
-	// 受限会话、待改密、未设密三者之一才允许免旧密码设密；其余按主动改密路径走。
-	if scope != SessionScopeSetPasswordOnly && !mustChange && (hash.Valid && hash.String != "") {
-		return "", ErrConflict
-	}
-	// 会话创建后账户凭据已变（并发改密/重置）：本次授权过期，拒绝继续。
-	if sessVer != userVer {
-		return "", ErrConflict
-	}
-	res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`),
-		passwordHash, nowISO(), false, owner)
-	if err != nil {
-		return "", err
-	}
-	if n, _ := res.RowsAffected(); n == 0 {
-		return "", ErrNotFound
-	}
-	if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`),
-		LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil {
-		return "", err
-	}
-	if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil {
-		return "", err
-	}
-	entrance := "must-change"
-	if scope == SessionScopeSetPasswordOnly {
-		entrance = "recovery"
-	}
-	if err := s.insertAccountEvent(tx, owner, entrance, owner, "password_set", entrance); err != nil {
-		return "", err
-	}
-	return owner, s.commitAccountMutation(tx, AccountWrite{Owner: owner, Actor: owner, Action: "account.set_password"})
+	return lockOwner, nil
 }
 
 // ChangeUserPasswordActive 主动改密单事务：校验当前密码 → 更新（带认证版本乐观
 // 检查）→ bump auth_version → 吊销其他会话（保留当前）→ 审计。中途失败无部分成功。
 func (s *Store) ChangeUserPasswordActive(owner, current, passwordHash, keepToken string) error {
-	tx, err := s.db.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
-	var hash string
-	var ver int
-	var mustChange bool
-	err = tx.QueryRow(s.q(`SELECT password_hash,auth_version,must_change_password FROM users WHERE owner=?`), owner).
-		Scan(&hash, &ver, &mustChange)
-	if errors.Is(err, sql.ErrNoRows) {
-		return ErrNotFound
-	}
-	if err != nil {
-		return err
-	}
-	if mustChange || hash == "" {
-		// 强制态必须走受限设密路径，不在此处用旧密码换新密码。
-		return ErrConflict
-	}
-	if !VerifyPassword(hash, current) {
-		return ErrBadCredentials
-	}
-	res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=? AND auth_version=?`),
-		passwordHash, nowISO(), false, owner, ver)
-	if err != nil {
-		return err
-	}
-	if n, _ := res.RowsAffected(); n == 0 {
-		return ErrConflict // 并发改密：版本已前进
-	}
-	if keepToken != "" {
-		if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=? AND token_hash<>?`), owner, hashKey(keepToken)); err != nil {
+	return s.accountTx(AccountWrite{Owner: owner, Actor: owner, Action: "account.change_password"}, func(tx *sql.Tx, _ string) error {
+		var hash string
+		var ver int
+		var mustChange bool
+		err := tx.QueryRow(s.q(`SELECT password_hash,auth_version,must_change_password FROM users WHERE owner=?`), owner).
+			Scan(&hash, &ver, &mustChange)
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		if err != nil {
 			return err
 		}
-	} else if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil {
-		return err
-	}
-	if err := s.insertAccountEvent(tx, owner, "settings", owner, "password_change", ""); err != nil {
-		return err
-	}
-	return s.commitAccountMutation(tx, AccountWrite{Owner: owner, Actor: owner, Action: "account.change_password"})
+		if mustChange || hash == "" {
+			return ErrConflict
+		}
+		if !VerifyPassword(hash, current) {
+			return ErrBadCredentials
+		}
+		res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=? AND auth_version=?`),
+			passwordHash, nowISO(), false, owner, ver)
+		if err != nil {
+			return err
+		}
+		if n, _ := res.RowsAffected(); n != 1 {
+			return ErrConflict
+		}
+		if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`),
+			LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil {
+			return err
+		}
+		if keepToken != "" {
+			if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=? AND token_hash<>?`), owner, hashKey(keepToken)); err != nil {
+				return err
+			}
+			if _, err := tx.Exec(s.q(`UPDATE sessions SET auth_version=? WHERE owner=? AND token_hash=?`),
+				ver+1, owner, hashKey(keepToken)); err != nil {
+				return err
+			}
+		} else if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil {
+			return err
+		}
+		return s.insertAccountEvent(tx, owner, "settings", owner, "password_change", "")
+	})
 }
 
 // AdminResetUserPassword 管理员/CLI 重置密码单事务：写 hash、must_change=true、
 // bump auth_version、作废未用登录 Key、吊销全部会话、审计（actor 为操作者）。
 func (s *Store) AdminResetUserPassword(owner, passwordHash, actor, entrance string) error {
-	tx, err := s.db.Begin()
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
-	res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`),
-		passwordHash, nowISO(), true, owner)
-	if err != nil {
-		return err
-	}
-	if n, _ := res.RowsAffected(); n == 0 {
-		return ErrNotFound
-	}
-	if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`),
-		LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil {
-		return err
-	}
-	if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil {
-		return err
-	}
 	if entrance == "" {
 		entrance = entranceOf(actor)
 	}
-	if err := s.insertAccountEvent(tx, actor, entrance, owner, "admin_password_reset", ""); err != nil {
-		return err
-	}
-	return s.commitAccountMutation(tx, AccountWrite{Owner: owner, Actor: actor, Action: "account.reset_password"})
+	return s.accountTx(AccountWrite{Owner: owner, Actor: actor, Action: "account.reset_password"}, func(tx *sql.Tx, _ string) error {
+		res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`),
+			passwordHash, nowISO(), true, owner)
+		if err != nil {
+			return err
+		}
+		if n, _ := res.RowsAffected(); n != 1 {
+			return ErrNotFound
+		}
+		if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`),
+			LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil {
+			return err
+		}
+		return s.insertAccountEvent(tx, actor, entrance, owner, "admin_password_reset", "")
+	})
 }
 
 // CountUnusedLoginKeys 未使用登录 Key 计数（list 展示）。