Fence final run settlement against account disable

Why: A disable could commit after the run manager checked account state but before its final business message was stored.

Evidence: SELF receipts /tmp/iam-u3-final-settle-sqlite.json, /tmp/iam-u3-final-settle-pg.json, /tmp/iam-u3-backend-settle-all.json, /tmp/iam-u3-backend-settle-pg.json, and /tmp/iam-u3-http-two-instance-pg.json.

Impact: The final run transaction locks and rechecks the account, rejects post-disable business output, and permits only an account-denial failure record. Auth-off implicit local ownership remains explicit.

Refs: IAM U3 r1 stages 4-5; Dev feature/iam-u3-account-directory-recovery

internal/app/app.go

@@ -452,7 +452,7 @@ func Run(opt Options) error {
 	mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{})
 	authOn := resolved.AuthEnabled()
 	effectiveAccount := accountEffectiveAuthorization(st, authOn)
-	mgr.SetAccountEffectiveAuthorization(effectiveAccount)
+	mgr.SetAccountEffectiveAuthorization(effectiveAccount, !authOn)
 	instanceID := opt.InstanceID
 	if instanceID == "" {
 		host, err := os.Hostname()
@@ -650,7 +650,9 @@ func Run(opt Options) error {
 		}
 	}
 	hostSrv.Dispatch = func(ctx context.Context, pluginID, owner, commandID string, params map[string]any) error {
-		if err := caps.accountGuard(owner); err != nil { return err }
+		if err := caps.accountGuard(owner); err != nil {
+			return err
+		}
 		resp := bus.Dispatch(ctx, cmdbus.Request{ID: commandID, Params: params, Entrance: "api", Actor: pluginID, Owner: owner})
 		if resp.Receipt.State == "rejected" || resp.Receipt.State == "unknown" {
 			return fmt.Errorf("%s", resp.Receipt.Reason)

internal/runmgr/runmgr.go

@@ -62,6 +62,7 @@ type Manager struct {
 	knowledgeProjectScope         func(context.Context, string, core.ID) (core.ID, error)
 	projectWorkAuthorization      func(context.Context, string, core.ID) (bool, error)
 	accountEffectiveAuthorization func(context.Context, string) (bool, error)
+	accountAllowImplicitLocal     bool
 	// commands EXT-14：命令即工具的投影与分发源（nil = 未接线，命令不进工具清单）。
 	commands CommandSource
 	indexer  core.DocumentIndexer // 可选索引入口（nil = store 整文档单块）
@@ -222,8 +223,11 @@ func (m *Manager) SetProjectWorkAuthorization(fn func(context.Context, string, c
 
 // SetAccountEffectiveAuthorization binds the current account-state guard. It is
 // independent of the optional domain project provider.
-func (m *Manager) SetAccountEffectiveAuthorization(fn func(context.Context, string) (bool, error)) {
+func (m *Manager) SetAccountEffectiveAuthorization(fn func(context.Context, string) (bool, error), allowImplicitLocal ...bool) {
 	m.accountEffectiveAuthorization = fn
+	if len(allowImplicitLocal) > 0 {
+		m.accountAllowImplicitLocal = allowImplicitLocal[0]
+	}
 }
 
 func (m *Manager) authorizeAccountEffective(ctx context.Context, owner string) error {
@@ -1696,7 +1700,11 @@ func (m *Manager) emit(conversationID, runID string, typ core.StreamEventType, p
 // settleRunWithFence（U5）：PG 档以本实例登记的 epoch 做围栏结算。
 func (m *Manager) settleRunWithFence(rc *runCtx, msg *core.Message) (updated, fenced bool, err error) {
 	if m.st.Dialect() != "postgres" {
-		updated, err = m.st.SettleRun(&rc.run, msg)
+		if m.accountEffectiveAuthorization != nil {
+			updated, err = m.st.SettleRunAccountChecked(&rc.run, msg, m.accountAllowImplicitLocal)
+		} else {
+			updated, err = m.st.SettleRun(&rc.run, msg)
+		}
 		return updated, false, err
 	}
 	m.execMu.Lock()
@@ -1708,6 +1716,9 @@ func (m *Manager) settleRunWithFence(rc *runCtx, msg *core.Message) (updated, fe
 			epoch = es.Epoch
 		}
 	}
+	if m.accountEffectiveAuthorization != nil {
+		return m.st.SettleRunFencedAccountChecked(&rc.run, msg, m.InstanceID(), epoch, m.accountAllowImplicitLocal)
+	}
 	return m.st.SettleRunFenced(&rc.run, msg, m.InstanceID(), epoch)
 }
 

internal/server/auth_e2e_test.go

@@ -36,7 +36,11 @@ func newAuthTestServer(t *testing.T) (*Server, *store.Store) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { st.Close() })
+	return newAuthTestServerOnStore(t, st, dir), st
+}
 
+func newAuthTestServerOnStore(t *testing.T, st *store.Store, dir string) *Server {
+	t.Helper()
 	b := bus.New()
 	reg := tools.NewRegistry()
 	reg.RegisterBuiltin()
@@ -48,7 +52,7 @@ func newAuthTestServer(t *testing.T) (*Server, *store.Store) {
 	srv := New(st, mgr, reg, profSvc, setSvc, ph)
 	srv.Auth = NewAuthenticator(st, true)
 	srv.Subscribe = func(cid string) (<-chan core.StreamEvent, func()) { return b.Subscribe(cid) }
-	return srv, st
+	return srv
 }
 
 func TestAuthMiddleware(t *testing.T) {

internal/server/recovery_multi_instance_test.go

@@ -0,0 +1,124 @@
+package server
+
+import (
+	"database/sql"
+	"encoding/json"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"agent-system-backend/internal/store"
+)
+
+func twinRecoveryStores(t *testing.T) (*store.Store, *store.Store) {
+	t.Helper()
+	dsn := os.Getenv("AGENT_PG_TEST_DSN")
+	if dsn == "" {
+		path := filepath.Join(t.TempDir(), "shared.db")
+		first, err := store.OpenSQLite(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		second, err := store.OpenSQLite(path)
+		if err != nil {
+			first.Close()
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { second.Close(); first.Close() })
+		return first, second
+	}
+	admin, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil {
+		t.Fatal(err)
+	}
+	schema := "t_" + strings.ToLower(store.NewID("recovery"))
+	if _, err := admin.Exec(`CREATE SCHEMA ` + schema); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`); admin.Close() })
+	isolate := store.AppendPGRuntimeParam(dsn, "search_path", schema+",public")
+	first, err := store.OpenPostgres(isolate, filepath.Join(t.TempDir(), "a-artifacts"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := store.OpenPostgres(isolate, filepath.Join(t.TempDir(), "b-artifacts"))
+	if err != nil {
+		first.Close()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { second.Close(); first.Close() })
+	return first, second
+}
+
+func TestRecoveryAcrossTwoHTTPInstancesAndIndependentPools(t *testing.T) {
+	first, second := twinRecoveryStores(t)
+	t.Logf("recovery HTTP dialect=%s two independent stores", first.Dialect())
+	hA := newAuthTestServerOnStore(t, first, t.TempDir()).Handler()
+	hB := newAuthTestServerOnStore(t, second, t.TempDir()).Handler()
+	adminCookie := bootAdmin(t, hA, first)
+	account, err := first.CreatePendingAccount("admin", "two-instance-owner", store.AccountProfile{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	activation, err := first.InitializePendingAccount(account.ID, account.StateVersion, account.InitializationGeneration)
+	if err != nil {
+		t.Fatal(err)
+	}
+	setup := do(hB, "POST", "/api/auth/login-key/redeem", `{"key":"`+activation+`"}`, "")
+	if setup.Code != http.StatusOK {
+		t.Fatalf("second instance activation redeem: %d %s", setup.Code, setup.Body.String())
+	}
+	if done := do(hA, "POST", "/api/auth/password", `{"next":"two-instance-initial"}`, cookieOf(setup)); done.Code != http.StatusOK {
+		t.Fatalf("first instance initial set: %d %s", done.Code, done.Body.String())
+	}
+	login := do(hA, "POST", "/api/auth/login", `{"owner":"two-instance-owner","password":"two-instance-initial"}`, "")
+	if login.Code != http.StatusOK {
+		t.Fatalf("first instance login: %d %s", login.Code, login.Body.String())
+	}
+	ownerCookie := cookieOf(login)
+	current, err := second.GetAccount(account.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	proposalResponse := do(hA, "POST", "/api/admin/users/"+account.ID+"/recovery-proposals",
+		`{"expectedStateVersion":`+jsonNumber(current.StateVersion)+`,"idempotencyKey":"two-instance-proposal"}`, adminCookie)
+	if proposalResponse.Code != http.StatusOK {
+		t.Fatalf("proposal: %d %s", proposalResponse.Code, proposalResponse.Body.String())
+	}
+	var proposal store.RecoveryProposal
+	if err := json.Unmarshal(proposalResponse.Body.Bytes(), &proposal); err != nil {
+		t.Fatal(err)
+	}
+	confirmation := do(hB, "POST", "/api/auth/recovery-proposals/"+proposal.ID+"/confirm", `{"expectedVersion":1}`, ownerCookie)
+	if confirmation.Code != http.StatusOK {
+		t.Fatalf("second instance confirm: %d %s", confirmation.Code, confirmation.Body.String())
+	}
+	var confirmed struct {
+		Grant string `json:"grant"`
+	}
+	if err := json.Unmarshal(confirmation.Body.Bytes(), &confirmed); err != nil || confirmed.Grant == "" {
+		t.Fatalf("grant response: %v", err)
+	}
+	redeemed := do(hA, "POST", "/api/auth/recovery-grants/redeem", `{"key":"`+confirmed.Grant+`"}`, "")
+	if redeemed.Code != http.StatusOK {
+		t.Fatalf("first instance grant redeem: %d %s", redeemed.Code, redeemed.Body.String())
+	}
+	if completed := do(hB, "POST", "/api/auth/password", `{"next":"two-instance-recovered"}`, cookieOf(redeemed)); completed.Code != http.StatusOK {
+		t.Fatalf("second instance recovery completion: %d %s", completed.Code, completed.Body.String())
+	}
+	if old := do(hB, "GET", "/api/me", "", ownerCookie); old.Code != http.StatusUnauthorized {
+		t.Fatalf("old proof session survived: %d", old.Code)
+	}
+	if fresh := do(hA, "POST", "/api/auth/login", `{"owner":"two-instance-owner","password":"two-instance-recovered"}`, ""); fresh.Code != http.StatusOK {
+		t.Fatalf("recovered password login: %d %s", fresh.Code, fresh.Body.String())
+	}
+	after, err := first.GetAccount(account.ID)
+	if err != nil || after.CredentialOrigin != store.CredentialOriginOwnerRecovery || after.ID != account.ID {
+		t.Fatalf("persistent identity or provenance changed: %+v %v", after, err)
+	}
+}

internal/store/account_settle_test.go

@@ -0,0 +1,62 @@
+package store
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	core "agent-system-core"
+)
+
+func TestAccountCheckedFinalSettlementCannotWriteAfterDisable(t *testing.T) {
+	st := openTestStore(t)
+	t.Logf("account final settlement dialect=%s", st.Dialect())
+	if err := st.CreateAccount("admin", "", "hash", true); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.EnsureUser(OwnerDefault, ""); err != nil {
+		t.Fatal(err)
+	}
+	conv := seedConversation(t, st)
+	run := seedRun(t, st, conv, core.RunRunning)
+	local, err := st.GetAccountByOwner(OwnerDefault)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.SetAccountEnabled("admin", local.ID, local.StateVersion, false); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now().UTC()
+	run.Status, run.Phase, run.EndedAt = core.RunCompleted, core.PhaseDone, &now
+	message := core.Message{ConversationID: conv.ID, RunID: run.ID, Role: core.RoleAssistant,
+		Parts: []core.Part{{Type: core.PartText, Text: "must not persist after disable"}}}
+	settle := func() error {
+		if st.Dialect() == "postgres" {
+			_, _, err := st.SettleRunFencedAccountChecked(&run, &message, "test-instance", 0, true)
+			return err
+		}
+		_, err := st.SettleRunAccountChecked(&run, &message, true)
+		return err
+	}
+	if err := settle(); !errors.Is(err, ErrAccountForbidden) {
+		t.Fatalf("disabled account completed business output: %v", err)
+	}
+	got, err := st.GetRunByID(run.ID)
+	if err != nil || got.Status != core.RunRunning {
+		t.Fatalf("failed transaction changed run state: %+v %v", got, err)
+	}
+	messages, err := st.ListMessages(conv.ID)
+	if err != nil || len(messages) != 0 {
+		t.Fatalf("failed transaction wrote assistant message: %+v %v", messages, err)
+	}
+	run.Status = core.RunFailed
+	run.Error = &core.ErrorPart{Code: "run.account_inactive", Message: "account disabled"}
+	message.Parts = []core.Part{{Type: core.PartError, Err: run.Error}}
+	if err := settle(); err != nil {
+		t.Fatalf("minimal account-denial failure could not settle: %v", err)
+	}
+	got, err = st.GetRunByID(run.ID)
+	if err != nil || got.Status != core.RunFailed {
+		t.Fatalf("denial state not persisted: %+v %v", got, err)
+	}
+}

internal/store/recovery.go

@@ -126,6 +126,36 @@ func (s *Store) ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error) {
 // (instance,epoch) 仍是该会话的当前执行权，被接管则拒写（fenced=true，run
 // 不变）。无租约行（未启用执行所有权）恒放行，语义与 SettleRun 一致。
 func (s *Store) SettleRunFenced(r *core.Run, msg *core.Message, instanceID string, epoch int64) (updated, fenced bool, err error) {
+	return s.settleRunRetried(r, msg, &fenceRef{instance: instanceID, epoch: epoch, conversation: string(r.ConversationID)}, nil)
+}
+
+// SettleRunFencedAccountChecked adds a transaction-time account check to the
+// execution fence. A disable committing first prevents a later business
+// message from reaching the authoritative message tree.
+func (s *Store) SettleRunFencedAccountChecked(r *core.Run, msg *core.Message, instanceID string, epoch int64, allowImplicitLocal bool) (bool, bool, error) {
+	return s.settleRunRetried(r, msg, &fenceRef{instance: instanceID, epoch: epoch, conversation: string(r.ConversationID)},
+		&accountSettleCheck{allowImplicitLocal: allowImplicitLocal})
+}
+
+// fenceRef 执行围栏引用（nil = 不校验）。
+type fenceRef struct {
+	instance, conversation string
+	epoch                  int64
+}
+
+type accountSettleCheck struct{ allowImplicitLocal bool }
+
+func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err error) {
+	updated, _, err = s.settleRunRetried(r, msg, nil, nil)
+	return updated, err
+}
+
+func (s *Store) SettleRunAccountChecked(r *core.Run, msg *core.Message, allowImplicitLocal bool) (bool, error) {
+	updated, _, err := s.settleRunRetried(r, msg, nil, &accountSettleCheck{allowImplicitLocal: allowImplicitLocal})
+	return updated, err
+}
+
+func (s *Store) settleRunRetried(r *core.Run, msg *core.Message, fence *fenceRef, account *accountSettleCheck) (updated, fenced bool, err error) {
 	s.settleMu.Lock()
 	fault := s.settleFault
 	s.settleMu.Unlock()
@@ -137,7 +167,7 @@ func (s *Store) SettleRunFenced(r *core.Run, msg *core.Message, instanceID strin
 	}
 	var last error
 	for attempt := 0; attempt < 8; attempt++ {
-		updated, fenced, last = s.settleRunOnce(r, msg, &fenceRef{instance: instanceID, epoch: epoch, conversation: string(r.ConversationID)})
+		updated, fenced, last = s.settleRunOnce(r, msg, fence, account)
 		if last == nil {
 			return updated, fenced, nil
 		}
@@ -152,46 +182,46 @@ func (s *Store) SettleRunFenced(r *core.Run, msg *core.Message, instanceID strin
 	return false, false, last
 }
 
-// fenceRef 执行围栏引用（nil = 不校验）。
-type fenceRef struct {
-	instance, conversation string
-	epoch                  int64
-}
-
-func (s *Store) SettleRun(r *core.Run, msg *core.Message) (updated bool, err error) {
-	s.settleMu.Lock()
-	fault := s.settleFault
-	s.settleMu.Unlock()
-	if fault != nil {
-		return false, fault
-	}
-	if msg != nil && msg.ID == "" {
-		msg.ID = newID("msg")
-	}
-	var last error
-	for attempt := 0; attempt < 8; attempt++ {
-		var fenced bool
-		updated, fenced, last = s.settleRunOnce(r, msg, nil)
-		if last == nil {
-			return updated && !fenced, nil
-		}
-		if !isUniqueViolation(last) {
-			return false, last
-		}
-		time.Sleep(time.Duration(attempt) * time.Millisecond)
-	}
-	if last == nil {
-		last = fmt.Errorf("store: settle retries exhausted (run %s)", r.ID)
-	}
-	return false, last
-}
-
-func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef) (bool, bool, error) {
+func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef, account *accountSettleCheck) (bool, bool, error) {
 	tx, err := s.db.Begin()
 	if err != nil {
 		return false, false, err
 	}
 	defer tx.Rollback()
+	if account != nil {
+		query := s.q(`SELECT enabled FROM users WHERE owner=?`)
+		if s.Dialect() == "postgres" {
+			query += ` FOR UPDATE`
+		}
+		var enabled bool
+		err := tx.QueryRow(query, r.Owner).Scan(&enabled)
+		if errors.Is(err, sql.ErrNoRows) {
+			if account.allowImplicitLocal && r.Owner == OwnerDefault {
+				enabled, err = true, nil
+			} else {
+				return false, false, ErrAccountForbidden
+			}
+		}
+		if err != nil {
+			return false, false, err
+		}
+		if !enabled {
+			// An already identified account-denial failure may persist a minimal
+			// error state; it contains no assistant content or business output.
+			allowedFailure := r.Status == core.RunFailed && r.Error != nil && r.Error.Code == "run.account_inactive"
+			if msg != nil {
+				for _, part := range msg.Parts {
+					if part.Type != core.PartError || part.Err == nil || part.Err.Code != "run.account_inactive" {
+						allowedFailure = false
+						break
+					}
+				}
+			}
+			if !allowedFailure {
+				return false, false, ErrAccountForbidden
+			}
+		}
+	}
 	if err := s.dialect.lockConversation(tx, string(r.ConversationID)); err != nil {
 		return false, false, err
 	}