diff --git a/internal/app/app.go b/internal/app/app.go index 8f27dc8..b7d96f3 100644 --- a/internal/app/app.go +++ b/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) diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 0833ca5..0f72efb 100644 --- a/internal/runmgr/runmgr.go +++ b/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) } diff --git a/internal/server/auth_e2e_test.go b/internal/server/auth_e2e_test.go index b9ea234..fa09148 100644 --- a/internal/server/auth_e2e_test.go +++ b/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) { diff --git a/internal/server/recovery_multi_instance_test.go b/internal/server/recovery_multi_instance_test.go new file mode 100644 index 0000000..f8dc950 --- /dev/null +++ b/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) + } +} diff --git a/internal/store/account_settle_test.go b/internal/store/account_settle_test.go new file mode 100644 index 0000000..dc89175 --- /dev/null +++ b/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) + } +} diff --git a/internal/store/recovery.go b/internal/store/recovery.go index 0ce0b49..08b0354 100644 --- a/internal/store/recovery.go +++ b/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 }