From b8b3493bb7019833c7894422ff148f488a3d39b9 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 17 Oct 2026 10:00:56 +0000 Subject: [PATCH 1/9] Enable admin password creation and reset --- internal/server/account_contract_test.go | 96 ++++++++++++++++++++++- internal/server/accounts.go | 55 +++++++++++-- internal/server/auth_e2e_test.go | 28 +++++-- internal/server/recovery_contract_test.go | 3 +- internal/store/account_model.go | 47 +++++++++++ 5 files changed, 212 insertions(+), 17 deletions(-) diff --git a/internal/server/account_contract_test.go b/internal/server/account_contract_test.go index e0d11f6..429d519 100644 --- a/internal/server/account_contract_test.go +++ b/internal/server/account_contract_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/http" "net/url" + "strings" "testing" "agent-system-backend/internal/store" @@ -113,7 +114,6 @@ func TestAccountHTTPPendingCreateProfileAndDisable(t *testing.T) { h := srv.Handler() adminCookie := bootAdmin(t, h, st) for _, body := range []string{ - `{"owner":"blocked-password","password":"old-pass-9"}`, `{"owner":"blocked-admin","isAdmin":true}`, } { if rec := do(h, "POST", "/api/admin/users", body, adminCookie); rec.Code != http.StatusBadRequest { @@ -152,3 +152,97 @@ func TestAccountHTTPPendingCreateProfileAndDisable(t *testing.T) { t.Fatalf("state/profile after disable: %+v %v", account, err) } } + +func TestAdminPasswordCreateAndResetHTTP(t *testing.T) { + srv, st := newAuthTestServer(t) + h := srv.Handler() + admin := bootAdmin(t, h, st) + + created := do(h, "POST", "/api/admin/users", `{"owner":"alice","profile":{"name":"爱丽丝"},"password":"alice-initial-9"}`, admin) + if created.Code != http.StatusOK { + t.Fatalf("password create: %d %s", created.Code, created.Body.String()) + } + if strings.Contains(created.Body.String(), "alice-initial-9") { + t.Fatal("supplied password was echoed") + } + var createdBody struct { + UserID string `json:"userId"` + PasswordSet bool `json:"passwordSet"` + MustChangePassword bool `json:"mustChangePassword"` + InitialPassword string `json:"initialPassword"` + } + if err := json.Unmarshal(created.Body.Bytes(), &createdBody); err != nil { + t.Fatal(err) + } + if createdBody.UserID == "" || !createdBody.PasswordSet || !createdBody.MustChangePassword || createdBody.InitialPassword != "" { + t.Fatalf("supplied password projection: %+v", createdBody) + } + account, err := st.GetAccountByOwner("alice") + if err != nil || !account.HasPassword || !account.MustChangePassword || account.InitializationState != store.InitializationInitialized { + t.Fatalf("stored account state: %+v %v", account, err) + } + if state, ok := st.VerifyLogin("alice", "alice-initial-9"); !ok || !state.MustChangePassword { + t.Fatalf("supplied password login: ok=%v state=%+v", ok, state) + } + + generated := do(h, "POST", "/api/admin/users", `{"owner":"bob"}`, admin) + if generated.Code != http.StatusOK { + t.Fatalf("generated password create: %d %s", generated.Code, generated.Body.String()) + } + var generatedBody struct { + InitialPassword string `json:"initialPassword"` + MustChangePassword bool `json:"mustChangePassword"` + } + if err := json.Unmarshal(generated.Body.Bytes(), &generatedBody); err != nil { + t.Fatal(err) + } + if generatedBody.InitialPassword == "" || !generatedBody.MustChangePassword { + t.Fatalf("generated password projection: %+v", generatedBody) + } + if state, ok := st.VerifyLogin("bob", generatedBody.InitialPassword); !ok || !state.MustChangePassword { + t.Fatalf("generated password login: ok=%v state=%+v", ok, state) + } + + login := do(h, "POST", "/api/auth/login", `{"owner":"alice","password":"alice-initial-9"}`, "") + if login.Code != http.StatusOK { + t.Fatalf("alice login: %d %s", login.Code, login.Body.String()) + } + aliceSession := cookieOf(login) + key, _, err := st.IssueLoginKey("alice", store.LoginKeyPurposeRecover, "test", 0) + if err != nil { + t.Fatal(err) + } + if rejected := do(h, "POST", "/api/admin/users/alice/password", `{"password":"client-supplied"}`, admin); rejected.Code != http.StatusBadRequest { + t.Fatalf("client reset password accepted: %d %s", rejected.Code, rejected.Body.String()) + } + if _, ok := st.VerifyLogin("alice", "alice-initial-9"); !ok { + t.Fatal("rejected reset changed the password") + } + + reset := do(h, "POST", "/api/admin/users/alice/password", `{}`, admin) + if reset.Code != http.StatusOK || reset.Header().Get("Cache-Control") != "no-store" { + t.Fatalf("admin reset: %d %s", reset.Code, reset.Body.String()) + } + var resetBody struct { + TemporaryPassword string `json:"temporaryPassword"` + MustChangePassword bool `json:"mustChangePassword"` + } + if err := json.Unmarshal(reset.Body.Bytes(), &resetBody); err != nil { + t.Fatal(err) + } + if resetBody.TemporaryPassword == "" || resetBody.TemporaryPassword == "alice-initial-9" || !resetBody.MustChangePassword { + t.Fatalf("reset projection: %+v", resetBody) + } + if rec := do(h, "GET", "/api/me", "", aliceSession); rec.Code != http.StatusUnauthorized { + t.Fatalf("reset did not revoke session: %d %s", rec.Code, rec.Body.String()) + } + if _, ok := st.VerifyLogin("alice", "alice-initial-9"); ok { + t.Fatal("old password remained usable after reset") + } + if state, ok := st.VerifyLogin("alice", resetBody.TemporaryPassword); !ok || !state.MustChangePassword { + t.Fatalf("temporary password login: ok=%v state=%+v", ok, state) + } + if _, _, err := st.RedeemLoginKey(key); err == nil { + t.Fatal("admin reset left unused login key redeemable") + } +} diff --git a/internal/server/accounts.go b/internal/server/accounts.go index 1b0b734..8e68ddc 100644 --- a/internal/server/accounts.go +++ b/internal/server/accounts.go @@ -324,8 +324,9 @@ func (s *Server) createAdminUser(w http.ResponseWriter, r *http.Request) { return } var req struct { - Owner string `json:"owner"` - Profile store.AccountProfile `json:"profile"` + Owner string `json:"owner"` + Profile store.AccountProfile `json:"profile"` + Password *string `json:"password"` } if err := decodeStrictJSON(r, &req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) @@ -336,12 +337,31 @@ func (s *Server) createAdminUser(w http.ResponseWriter, r *http.Request) { http.Error(w, "owner required (no spaces or slashes)", http.StatusBadRequest) return } - account, err := s.st.CreatePendingAccount(actor, req.Owner, req.Profile) + password := RandomPassword() + generated := req.Password == nil + if req.Password != nil { + password = *req.Password + if err := validPassword(password); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + } + hash, err := store.HashPassword(password) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + account, err := s.st.CreateAdminPasswordAccount(actor, req.Owner, req.Profile, hash) if err != nil { writeStoreErr(w, err) return } - writeJSON(w, http.StatusOK, accountProfileJSON(account)) + out := accountProfileJSON(account) + if generated { + w.Header().Set("Cache-Control", "no-store") + out["initialPassword"] = password + } + writeJSON(w, http.StatusOK, out) } func accountProfileJSON(account store.Account) map[string]any { @@ -349,7 +369,8 @@ func accountProfileJSON(account store.Account) map[string]any { "userId": account.ID, "owner": account.Owner, "profile": account.Profile, "profileVersion": account.ProfileVersion, "enabled": account.Enabled, "stateVersion": account.StateVersion, "isAdmin": account.IsAdmin, - "passwordSet": account.HasPassword, "initializationState": account.InitializationState, + "passwordSet": account.HasPassword, "mustChangePassword": account.MustChangePassword, + "initializationState": account.InitializationState, "initializationGeneration": account.InitializationGeneration, } } @@ -466,9 +487,29 @@ func (s *Server) setAdminAccountState(w http.ResponseWriter, r *http.Request) { // 走签发一次性登录 Key）。单事务：must_change=1、bump auth_version、作废未用 // 登录 Key、吊销全部会话、审计。 func (s *Server) resetAdminUserPassword(w http.ResponseWriter, r *http.Request) { - if _, ok := s.requireAdmin(w, r); ok { - http.Error(w, "RECOVERY_CONTRACT_REPLACED: propose recovery instead", http.StatusGone) + actor, ok := s.requireAdmin(w, r) + if !ok { + return } + var req struct{} + if err := decodeStrictJSON(r, &req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + password := RandomPassword() + hash, err := store.HashPassword(password) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if err := s.st.AdminResetUserPassword(r.PathValue("owner"), hash, actor, "admin-http"); err != nil { + writeStoreErr(w, err) + return + } + w.Header().Set("Cache-Control", "no-store") + writeJSON(w, http.StatusOK, map[string]any{ + "owner": r.PathValue("owner"), "temporaryPassword": password, "mustChangePassword": true, + }) } // ---- 管理员：一次性登录 Key（恢复流程主入口）---- diff --git a/internal/server/auth_e2e_test.go b/internal/server/auth_e2e_test.go index 91d2a22..c3c2e34 100644 --- a/internal/server/auth_e2e_test.go +++ b/internal/server/auth_e2e_test.go @@ -423,19 +423,33 @@ func TestAdminAccountFlow(t *testing.T) { t.Fatalf("non-admin create must 403: %d", rec.Code) } - // 旧管理员重置入口关闭，不能动 alice 的密码与当前会话。 - if rec := session("POST", "/api/admin/users/alice/password", `{"password":"reset-init-02"}`); rec.Code != http.StatusGone { - t.Fatalf("admin reset must be closed: %d %s", rec.Code, rec.Body.String()) + // 管理员重置总是生成临时密码，销毁 alice 的旧会话并强制首登改密。 + reset := session("POST", "/api/admin/users/alice/password", `{}`) + if reset.Code != http.StatusOK || reset.Header().Get("Cache-Control") != "no-store" { + t.Fatalf("admin reset: %d %s", reset.Code, reset.Body.String()) } - if rec := alice("GET", "/api/me", ""); rec.Code != http.StatusOK { - t.Fatalf("closed reset must preserve alice session: %d", rec.Code) + var resetBody struct { + TemporaryPassword string `json:"temporaryPassword"` + } + if err := json.Unmarshal(reset.Body.Bytes(), &resetBody); err != nil || resetBody.TemporaryPassword == "" { + t.Fatalf("admin reset response: %+v %v", resetBody, err) + } + if rec := alice("GET", "/api/me", ""); rec.Code != http.StatusUnauthorized { + t.Fatalf("reset must revoke alice session: %d", rec.Code) } req = httptest.NewRequest("POST", "/api/auth/login", strings.NewReader(`{"owner":"alice","password":"alice-final-9"}`)) req.Header.Set("Content-Type", "application/json") rec = httptest.NewRecorder() h.ServeHTTP(rec, req) - if rec.Code != http.StatusOK || strings.Contains(rec.Body.String(), `"mustChangePassword":true`) { - t.Fatalf("closed reset must preserve normal login: %d %s", rec.Code, rec.Body.String()) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("old password remained usable after reset: %d %s", rec.Code, rec.Body.String()) + } + req = httptest.NewRequest("POST", "/api/auth/login", strings.NewReader(`{"owner":"alice","password":"`+resetBody.TemporaryPassword+`"}`)) + req.Header.Set("Content-Type", "application/json") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), `"mustChangePassword":true`) { + t.Fatalf("temporary password login: %d %s", rec.Code, rec.Body.String()) } // API key 通道兼容：密钥登录照旧（并具备其 owner 身份）。 diff --git a/internal/server/recovery_contract_test.go b/internal/server/recovery_contract_test.go index ad983bf..9918ffc 100644 --- a/internal/server/recovery_contract_test.go +++ b/internal/server/recovery_contract_test.go @@ -99,7 +99,7 @@ func TestOwnerConfirmedRecoveryHTTPKeepsGrantFromAdministrator(t *testing.T) { func jsonNumber(v int64) string { return strconv.FormatInt(v, 10) } -func TestLegacyAdministratorRecoveryEntrancesAreClosed(t *testing.T) { +func TestLegacyAdministratorLoginKeyIssuanceEntrancesAreClosed(t *testing.T) { srv, st := newAuthTestServer(t) h := srv.Handler() admin := bootAdmin(t, h, st) @@ -114,7 +114,6 @@ func TestLegacyAdministratorRecoveryEntrancesAreClosed(t *testing.T) { t.Fatal(err) } for _, route := range []struct{ path, body string }{ - {"/api/admin/users/victim/password", `{"password":"attacker-pass"}`}, {"/api/admin/users/victim/login-keys", `{"purpose":"recover"}`}, {"/api/admin/users/victim/login-keys", `{"purpose":"activate"}`}, } { diff --git a/internal/store/account_model.go b/internal/store/account_model.go index 7c81fb6..cd945c9 100644 --- a/internal/store/account_model.go +++ b/internal/store/account_model.go @@ -299,6 +299,53 @@ func (s *Store) CreatePendingAccount(actorOwner, owner string, profile AccountPr return a, nil } +// CreateAdminPasswordAccount creates an ordinary user with an administrator- +// supplied password hash. The account row, hash, and audit entries commit +// together so a failed write cannot leave a credential without its account. +func (s *Store) CreateAdminPasswordAccount(actorOwner, owner string, profile AccountProfile, passwordHash string) (Account, error) { + if passwordHash == "" { + return Account{}, ErrConflict + } + tx, err := s.db.Begin() + if err != nil { + return Account{}, err + } + defer tx.Rollback() + locked, err := s.lockAccountUsers(tx, actorOwner) + if errors.Is(err, ErrNotFound) { + return Account{}, ErrAccountForbidden + } + if err != nil { + return Account{}, err + } + actor := locked[actorOwner] + if !actor.Enabled || !actor.IsAdmin { + return Account{}, ErrAccountForbidden + } + a, err := s.insertPendingAccountTx(tx, owner, profile) + if err != nil { + return Account{}, err + } + if _, err := tx.Exec(s.q(`UPDATE users SET password_hash=?,password_set_at=?,must_change_password=?,credential_origin=?,initialization_state=? WHERE id=?`), + passwordHash, nowISO(), true, CredentialOriginAdminPassword, InitializationInitialized, a.ID); err != nil { + return Account{}, err + } + a, err = s.accountByOwner(tx, owner) + if err != nil { + return Account{}, err + } + if err := s.insertAccountEvent(tx, actorOwner, "account-management", owner, "account_created_password", a.ID); err != nil { + return Account{}, err + } + if err := s.auditAccountMutation(tx, owner, actorOwner, "account.create_password"); err != nil { + return Account{}, err + } + if err := tx.Commit(); err != nil { + return Account{}, err + } + return a, nil +} + // insertPendingAccountTx lets the trusted local initializer create and grant in // one transaction without impersonating an administrator. The caller owns its // authorization, audit and commit; this helper always inserts an ordinary user. From 5f65059bc28196c9b2621bf90c80b185146a0006 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 17 Oct 2026 11:56:33 +0000 Subject: [PATCH 2/9] Clarify admin reset contract --- internal/server/accounts.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/internal/server/accounts.go b/internal/server/accounts.go index 8e68ddc..d7971fe 100644 --- a/internal/server/accounts.go +++ b/internal/server/accounts.go @@ -483,9 +483,8 @@ func (s *Server) setAdminAccountState(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, accountProfileJSON(account)) } -// resetAdminUserPassword：管理员直接重置密码（兼容保留的既有入口；常规恢复 -// 走签发一次性登录 Key）。单事务：must_change=1、bump auth_version、作废未用 -// 登录 Key、吊销全部会话、审计。 +// resetAdminUserPassword：管理员生成随机临时密码。单事务：must_change=1、bump +// auth_version、作废未用登录 Key、吊销全部会话、审计。 func (s *Server) resetAdminUserPassword(w http.ResponseWriter, r *http.Request) { actor, ok := s.requireAdmin(w, r) if !ok { From b00e1924407989e52350ecdfa17f16640815ed89 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 15 Oct 2026 14:43:13 +0000 Subject: [PATCH 3/9] Execute the managed interpreter through its real path Why: the sandbox hides /home and only remounts the resolved Python prefix, so executing python/current/bin/python3 failed before the base layer could be imported. Evidence: go test ./internal/tools/ -run TestResolveInterpreterFollowsManagedSymlink passed, and env doctor against the built base layer printed doctor ok (SELF). Impact: run_python and env doctor follow the symlink into the mounted prefix or /usr/bin/python3. The layer and the prefix mount stay the same. Refs: Agent-System-Backend PR 207 Co-authored-by: Cursor (cherry picked from commit 28544cfe5da90e97474c42244ad8796e8029eed0) --- cmd/agent-system/env.go | 4 ++++ internal/tools/exec_interpreter.go | 8 +++++++- internal/tools/exec_interpreter_test.go | 22 ++++++++++++++++++++++ 3 files changed, 33 insertions(+), 1 deletion(-) diff --git a/cmd/agent-system/env.go b/cmd/agent-system/env.go index 818d51d..845169b 100644 --- a/cmd/agent-system/env.go +++ b/cmd/agent-system/env.go @@ -6,6 +6,7 @@ import ( "flag" "fmt" "os" + "path/filepath" "time" "agent-system-backend/internal/execenv" @@ -137,6 +138,9 @@ func runDoctor(ctx context.Context, root *execenv.Root, py string, sandbox bool) if prefix := root.PythonPrefix(); prefix != "" { mounts = append(mounts, prefix) } + if resolved, err := filepath.EvalSymlinks(py); err == nil { + py = resolved + } if err := execenv.Doctor(ctx, py, pyPath, nil, sandbox, mounts); err != nil { fmt.Fprintln(os.Stderr, "env doctor:", err) return 1 diff --git a/internal/tools/exec_interpreter.go b/internal/tools/exec_interpreter.go index 55f2917..d44aa9f 100644 --- a/internal/tools/exec_interpreter.go +++ b/internal/tools/exec_interpreter.go @@ -203,7 +203,13 @@ func resolveInterpreter(name string) (string, error) { case "python3": if execPythonBin != "" { if st, err := os.Stat(execPythonBin); err == nil && !st.IsDir() { - return execPythonBin, nil + // current/bin/python3 是符号链接。沙箱用 tmpfs 盖住 /home 后只挂真实前缀， + // 执行未解析的 current 路径会找不到文件。 + resolved, err := filepath.EvalSymlinks(execPythonBin) + if err != nil { + return "", err + } + return resolved, nil } return "", fmt.Errorf("python3: configured interpreter missing: %s", execPythonBin) } diff --git a/internal/tools/exec_interpreter_test.go b/internal/tools/exec_interpreter_test.go index da0aa29..90b4364 100644 --- a/internal/tools/exec_interpreter_test.go +++ b/internal/tools/exec_interpreter_test.go @@ -213,6 +213,28 @@ func TestExecEnvIncludesSkillsRoot(t *testing.T) { } } +func TestResolveInterpreterFollowsManagedSymlink(t *testing.T) { + prev := execPythonBin + t.Cleanup(func() { execPythonBin = prev }) + dir := t.TempDir() + link := filepath.Join(dir, "python3") + if err := os.Symlink("/usr/bin/python3", link); err != nil { + t.Fatal(err) + } + execPythonBin = link + got, err := resolveInterpreter("python3") + if err != nil { + t.Fatal(err) + } + want, err := filepath.EvalSymlinks("/usr/bin/python3") + if err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("resolved %q, want %q", got, want) + } +} + func TestInterpreterExecConfigForKernel(t *testing.T) { cfg := InterpreterExecConfig() if cfg == nil { From 7b379d28e74fc695dcda4750ad64f223d41ec5af Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Sun, 18 Oct 2026 09:20:03 +0000 Subject: [PATCH 4/9] agent-exec transport d2c63b4a83821a5716997a28174233b1/d6e403e1e778 --- cmd/agent-system/main.go | 9 +- cmd/agent-system/migrate.go | 30 ++ cmd/storage-migrate/main.go | 2 + internal/app/app.go | 47 +-- internal/app/guard_test.go | 201 ++++++++++++ internal/app/service_store.go | 111 +++++++ internal/instanceguard/guard.go | 199 ++++++++++++ internal/instanceguard/guard_other.go | 18 ++ internal/instanceguard/guard_test.go | 406 +++++++++++++++++++++++++ internal/instanceguard/guard_unix.go | 201 ++++++++++++ internal/store/durability.go | 67 ++++ internal/store/durability_test.go | 95 ++++++ internal/store/durability_unix_test.go | 35 +++ internal/store/instance_guard.go | 47 +++ internal/store/migrate.go | 28 ++ internal/store/migrate_compat.go | 1 + internal/store/store.go | 7 +- 17 files changed, 1459 insertions(+), 45 deletions(-) create mode 100644 cmd/agent-system/migrate.go create mode 100644 internal/app/guard_test.go create mode 100644 internal/app/service_store.go create mode 100644 internal/instanceguard/guard.go create mode 100644 internal/instanceguard/guard_other.go create mode 100644 internal/instanceguard/guard_test.go create mode 100644 internal/instanceguard/guard_unix.go create mode 100644 internal/store/durability.go create mode 100644 internal/store/durability_test.go create mode 100644 internal/store/durability_unix_test.go create mode 100644 internal/store/instance_guard.go diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 52e9428..f5f43fc 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -48,9 +48,11 @@ func main() { os.Exit(runCapabilities(os.Args[2:])) case "env": os.Exit(runEnv(os.Args[2:])) + case "migrate": + os.Exit(runMigrate(os.Args[2:])) default: fmt.Fprintf(os.Stderr, "agent-system: 未知子命令 %s\n", os.Args[1]) - fmt.Fprintln(os.Stderr, "用法：agent-system [参数]") + fmt.Fprintln(os.Stderr, "用法：agent-system [参数]") os.Exit(2) } } @@ -77,6 +79,7 @@ func main() { pgIdleSec := fs.Int("pg-conn-max-idle-sec", 0, "PG 空闲连接回收秒数（0=默认 300）") pgLifetimeSec := fs.Int("pg-conn-max-lifetime-sec", 0, "PG 连接最长生命周期秒数（0=不限）") pgQueryTimeoutSec := fs.Int("pg-query-timeout-sec", 0, "PG statement_timeout 秒数（0=不限；覆盖全部业务查询）") + adoptGuard := fs.Bool("adopt-instance-guard", false, "数据库被替换或锁文件被重建后，显式重写 SQLite 活动服务守卫") extension.RegisterCLIFlags(fs) fs.Parse(os.Args[1:]) handled, hookErr := extension.RunCLIHooks(extension.Options{ @@ -91,6 +94,7 @@ func main() { return } cliSet := visited(fs) + delete(cliSet, "adopt-instance-guard") resolved, err := config.Load(*cfgFile, os.LookupEnv, config.Config{ Addr: *addr, DataDir: *data, PluginsDir: *pluginsDir, Mock: *mock, @@ -141,7 +145,8 @@ func main() { DisableConvCardMirror: extOpt.DisableConvCardMirror, DisableExecTools: extOpt.DisableExecTools, ExecBackend: extOpt.ExecBackend, ExecDepsDir: resolved.ExecDepsDir, ExecWorkspace: resolved.ExecWorkspace, ExecEnvRoot: resolved.ExecEnvRoot, - Tier: resolved.Tier, + Tier: resolved.Tier, + AdoptInstanceGuard: *adoptGuard, PGPool: store.PoolOptions{ MaxOpenConn: resolved.PGMaxOpen, MaxIdleConn: resolved.PGMaxIdle, ConnMaxIdleTimeSec: resolved.PGConnMaxIdleSec, ConnMaxLifetimeSec: resolved.PGConnMaxLifetimeSec, diff --git a/cmd/agent-system/migrate.go b/cmd/agent-system/migrate.go new file mode 100644 index 0000000..21dd629 --- /dev/null +++ b/cmd/agent-system/migrate.go @@ -0,0 +1,30 @@ +package main + +import ( + "flag" + "fmt" + "os" + + "agent-system-backend/internal/app" +) + +func runMigrate(args []string) int { + fs := flag.NewFlagSet("migrate", flag.ExitOnError) + data := fs.String("data", "./data", "数据目录（SQLite 档）") + dsn := fs.String("dsn", "", "PostgreSQL DSN（空则读 AGENT_SYSTEM_DSN）") + adopt := fs.Bool("adopt-instance-guard", false, "数据库被替换或锁文件被重建后，显式重写守卫") + _ = fs.Parse(args) + storage := *dsn + if storage == "" { + storage = os.Getenv("AGENT_SYSTEM_DSN") + } + if err := app.Migrate(app.Options{ + DataDir: *data, + StorageDSN: storage, + AdoptInstanceGuard: *adopt, + }); err != nil { + fmt.Fprintln(os.Stderr, "agent-system:", err) + return 1 + } + return 0 +} diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index d1a8521..d4d0814 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -300,6 +300,8 @@ var copyOrder = []string{ "recovery_proposals", "recovery_proposal_proofs", // v35 分层 Python 环境（无外键；申请与会话绑定要随库迁走） "exec_env_requests", "exec_env_bindings", + // U8：活动服务锁的库内标记（无外键；迁移须保留以免目标库丢掉锁根） + "instance_guard", } const controlDDL = ` diff --git a/internal/app/app.go b/internal/app/app.go index a6edf18..e0ff13b 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -30,7 +30,6 @@ import ( "agent-native/runtime/telemetry" agenttools "agent-native/runtime/tools" nativeadp "agent-squad/adapters/native" - "agent-system-backend/internal/blobstore" "agent-system-backend/internal/bus" "agent-system-backend/internal/cardsvc" "agent-system-backend/internal/cmdbus" @@ -97,6 +96,8 @@ type Options struct { Tier string // PGPool PG 连接预算（KTD12；零值走默认）。仅 PG 档生效。 PGPool store.PoolOptions + // AdoptInstanceGuard 在数据库被替换或锁文件被重建后，显式重写守卫记录。 + AdoptInstanceGuard bool } // instanceTag 稳定执行身份：tier=team 时强制 host:addr（多实例互异）， @@ -175,49 +176,11 @@ func Run(opt Options) error { if opt.InsecureListen && !config.IsLoopbackListen(opt.Addr) && !resolved.AuthEnabled() { log.Printf("warning: listening on %s without authentication because insecureListen is set", opt.Addr) } - var st *store.Store - var err error - if opt.StorageDSN != "" { - artifacts := opt.ArtifactsDir - if artifacts == "" { - artifacts = filepath.Join(opt.DataDir, "artifacts") - } - // 共享工件（U5/KTD9）：配置了 S3 兼容对象存储时注入（多实例共享）； - // 未配置回落本地目录（单实例/共享卷）。凭据只经环境（不入库不入报告）。 - blobs, berr := blobstore.NewS3FromEnv() - if berr != nil { - return fmt.Errorf("blobstore s3: %w", berr) - } - if blobs != nil { - log.Printf("artifacts: shared via S3-compatible object storage (prefix %q)", blobs.Prefix) - } - st, err = store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, blobs) - if err != nil && opt.Tier == "team" { - // team 档拒绝就绪（R2）：PG 故障不回落 SQLite，错误必须明说。 - return fmt.Errorf("team 档打开 PostgreSQL 失败，拒绝就绪（不回落 SQLite）: %w", err) - } - } else { - st, err = store.OpenSQLite(filepath.Join(opt.DataDir, "agent-system.db")) - } + st, releaseStore, err := openServiceStore(opt) if err != nil { - return fmt.Errorf("open store: %w", err) - } - defer st.Close() - // U1 I3：svc 注入的服务实例身份（manifest.instance_id）与库内 marker 对齐； - // 未设置则跳过（兼容未接线旧调用方）。与 --instance（runs 进程标识）无关。 - if sysID := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")); sysID != "" { - dbref := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_DBREF")) - if dbref == "" { - if opt.StorageDSN != "" { - dbref = "pg" - } else { - dbref = "sqlite" - } - } - if err := st.VerifySystemIdentity(sysID, dbref); err != nil { - return fmt.Errorf("system identity: %w", err) - } + return err } + defer releaseStore() pool := st.Pool() if versions, verr := st.MigrationStatus(); verr == nil { log.Printf("storage tier: %s (dialect=%s pool=%d/%d idleSec=%d lifetimeSec=%d schema=v%d)", diff --git a/internal/app/guard_test.go b/internal/app/guard_test.go new file mode 100644 index 0000000..743c07e --- /dev/null +++ b/internal/app/guard_test.go @@ -0,0 +1,201 @@ +//go:build unix + +package app + +import ( + "errors" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "agent-system-backend/internal/instanceguard" + "agent-system-backend/internal/store" + core "agent-system-core" +) + +func TestSecondProcessRejectedBeforeOpen(t *testing.T) { + if os.Getenv("U8_APP_ROLE") == "hold" { + holdService(t) + return + } + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + cmd := exec.Command(os.Args[0], "-test.run=TestSecondProcessRejectedBeforeOpen") + cmd.Env = append(os.Environ(), "U8_APP_ROLE=hold", "U8_APP_DIR="+dir, "U8_APP_READY="+ready) + cmd.Dir = t.TempDir() + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + } + }) + waitReady(t, ready) + _, _, err := openServiceStore(Options{DataDir: dir, Addr: ":1"}) + if !errors.Is(err, instanceguard.ErrBusy) { + t.Fatalf("second open: %v", err) + } + if err := Migrate(Options{DataDir: dir}); !errors.Is(err, instanceguard.ErrBusy) { + t.Fatalf("migrate while held: %v", err) + } + cli, err := store.OpenExistingSQLite(filepath.Join(dir, "agent-system.db")) + if err != nil { + t.Fatal(err) + } + run, err := cli.GetRunByID("run_u8_hold") + if err != nil { + t.Fatal(err) + } + if run.Status != core.RunWaitingHITL { + t.Fatalf("run changed while second process was refused: %s", run.Status) + } + if _, err := cli.CreateProject("owner", "short"); err != nil { + t.Fatalf("account-style write while service holds the guard: %v", err) + } + if err := cli.Close(); err != nil { + t.Fatal(err) + } + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + _, _ = cmd.Process.Wait() + st, release, err := openServiceStore(Options{DataDir: dir, Addr: ":2"}) + if err != nil { + t.Fatal(err) + } + defer release() + run, err = st.GetRunByID("run_u8_hold") + if err != nil { + t.Fatal(err) + } + if run.Status != core.RunWaitingHITL { + t.Fatalf("restart reconciled a run the second process never opened: %s", run.Status) + } +} + +func TestCopiedBinaryStillSharesGuard(t *testing.T) { + if os.Getenv("U8_APP_ROLE") == "copy-hold" { + holdService(t) + return + } + bin, err := os.Executable() + if err != nil { + t.Fatal(err) + } + copyPath := filepath.Join(t.TempDir(), "agent-system-test-copy") + raw, err := os.ReadFile(bin) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(copyPath, raw, 0o700); err != nil { + t.Fatal(err) + } + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + cmd := exec.Command(copyPath, "-test.run=TestCopiedBinaryStillSharesGuard") + cmd.Env = append(os.Environ(), "U8_APP_ROLE=copy-hold", "U8_APP_DIR="+dir, "U8_APP_READY="+ready) + cmd.Dir = t.TempDir() + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + } + }) + waitReady(t, ready) + if _, _, err := openServiceStore(Options{DataDir: dir}); !errors.Is(err, instanceguard.ErrBusy) { + t.Fatalf("copied binary: %v", err) + } +} + +func holdService(t *testing.T) { + t.Helper() + dir := os.Getenv("U8_APP_DIR") + st, _, err := openServiceStore(Options{DataDir: dir}) + if err != nil { + t.Fatal(err) + } + run := &core.Run{ + ID: "run_u8_hold", ConversationID: "c", ProjectID: "p", Owner: "o", + Kind: core.RunConversation, Status: core.RunWaitingHITL, + } + if err := st.CreateRun(run); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(os.Getenv("U8_APP_READY"), []byte("ready"), 0o600); err != nil { + t.Fatal(err) + } + time.Sleep(30 * time.Second) +} + +func waitReady(t *testing.T, ready string) { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(ready); err == nil { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("child did not become ready") +} + +func TestAdoptDoesNotCreateMissingDatabase(t *testing.T) { + dir := t.TempDir() + _, release, err := openServiceStore(Options{DataDir: dir}) + if err != nil { + t.Fatal(err) + } + release() + db := filepath.Join(dir, "agent-system.db") + if err := os.Remove(db); err != nil { + t.Fatal(err) + } + _, _, err = openServiceStore(Options{DataDir: dir, AdoptInstanceGuard: true}) + if !errors.Is(err, instanceguard.ErrDBMissing) { + t.Fatalf("adopt missing: %v", err) + } + if _, err := os.Stat(db); !os.IsNotExist(err) { + t.Fatalf("adopt created a database: %v", err) + } +} + +func TestIdentityFailureDoesNotStick(t *testing.T) { + dir := t.TempDir() + t.Setenv("AGENT_SYSTEM_INSTANCE_ID", "wrong") + if _, _, err := openServiceStore(Options{DataDir: dir}); err == nil { + t.Fatal("missing identity row was accepted") + } + t.Setenv("AGENT_SYSTEM_INSTANCE_ID", "") + st, err := store.OpenSQLite(filepath.Join(dir, "agent-system.db")) + if err != nil { + t.Fatal(err) + } + marker, ok, err := st.LoadInstanceGuard() + if err != nil || ok || marker.InstanceID != "" { + t.Fatalf("failed start wrote a guard id: ok=%v id=%q err=%v", ok, marker.InstanceID, err) + } + if err := st.EnsureSystemIdentity("right", "sqlite"); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + t.Setenv("AGENT_SYSTEM_INSTANCE_ID", "right") + opened, release, err := openServiceStore(Options{DataDir: dir}) + if err != nil { + t.Fatal(err) + } + release() + _ = opened +} diff --git a/internal/app/service_store.go b/internal/app/service_store.go new file mode 100644 index 0000000..0d45a56 --- /dev/null +++ b/internal/app/service_store.go @@ -0,0 +1,111 @@ +package app + +import ( + "errors" + "fmt" + "log" + "os" + "path/filepath" + "strings" + + "agent-system-backend/internal/blobstore" + "agent-system-backend/internal/instanceguard" + "agent-system-backend/internal/store" +) + +// openServiceStore 打开活动服务使用的存储。 +// SQLite 档先取活动服务锁，再打开（打开即迁移），然后核对守卫标记。 +// 返回的函数先关存储再放锁。 +func openServiceStore(opt Options) (*store.Store, func(), error) { + nop := func() {} + if opt.StorageDSN != "" { + artifacts := opt.ArtifactsDir + if artifacts == "" { + artifacts = filepath.Join(opt.DataDir, "artifacts") + } + // 共享工件（U5/KTD9）：配置了 S3 兼容对象存储时注入（多实例共享）； + // 未配置回落本地目录（单实例/共享卷）。凭据只经环境（不入库不入报告）。 + blobs, berr := blobstore.NewS3FromEnv() + if berr != nil { + return nil, nop, fmt.Errorf("blobstore s3: %w", berr) + } + if blobs != nil { + log.Printf("artifacts: shared via S3-compatible object storage (prefix %q)", blobs.Prefix) + } + st, err := store.OpenPostgresWithOptions(opt.StorageDSN, artifacts, opt.PGPool, blobs) + if err != nil && opt.Tier == "team" { + // team 档拒绝就绪（R2）：PG 故障不回落 SQLite，错误必须明说。 + return nil, nop, fmt.Errorf("team 档打开 PostgreSQL 失败，拒绝就绪（不回落 SQLite）: %w", err) + } + if err != nil { + return nil, nop, fmt.Errorf("open store: %w", err) + } + if err := verifyConfiguredIdentity(st, "pg"); err != nil { + _ = st.Close() + return nil, nop, err + } + return st, func() { _ = st.Close() }, nil + } + + lease, err := instanceguard.Acquire(opt.DataDir) + if err != nil { + return nil, nop, fmt.Errorf("instance guard: %w", err) + } + releaseLease := func() { lease.Release() } + if err := lease.AutomaticRefusal(); err != nil { + adoptable := opt.AdoptInstanceGuard && (errors.Is(err, instanceguard.ErrDBReplaced) || errors.Is(err, instanceguard.ErrLockCorrupt)) + if !adoptable { + releaseLease() + return nil, nop, fmt.Errorf("instance guard: %w", err) + } + } + st, err := store.OpenSQLite(lease.DBPath()) + if err != nil { + releaseLease() + return nil, nop, fmt.Errorf("open store: %w", err) + } + if err := verifyConfiguredIdentity(st, "sqlite"); err != nil { + _ = st.Close() + releaseLease() + return nil, nop, err + } + if err := instanceguard.Bind(st, lease, instanceguard.BindOptions{ + Adopt: opt.AdoptInstanceGuard, + InstanceID: strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")), + }); err != nil { + _ = st.Close() + releaseLease() + return nil, nop, fmt.Errorf("instance guard: %w", err) + } + return st, func() { + _ = st.Close() + lease.Release() + }, nil +} + +// verifyConfiguredIdentity U1 I3：svc 注入的服务实例身份（manifest.instance_id）与库内 marker 对齐； +// 未设置则跳过（兼容未接线旧调用方）。与 --instance（runs 进程标识）无关。 +func verifyConfiguredIdentity(st *store.Store, defaultRef string) error { + sysID := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")) + if sysID == "" { + return nil + } + dbref := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_DBREF")) + if dbref == "" { + dbref = defaultRef + } + if err := st.VerifySystemIdentity(sysID, dbref); err != nil { + return fmt.Errorf("system identity: %w", err) + } + return nil +} + +// Migrate 执行显式迁移。SQLite 必须先取得活动服务锁，有服务占用则拒绝。 +func Migrate(opt Options) error { + _, release, err := openServiceStore(opt) + if err != nil { + return err + } + release() + return nil +} diff --git a/internal/instanceguard/guard.go b/internal/instanceguard/guard.go new file mode 100644 index 0000000..c086727 --- /dev/null +++ b/internal/instanceguard/guard.go @@ -0,0 +1,199 @@ +// Package instanceguard 强制同一 SQLite 数据文件只有一个活动服务。 +// 锁在打开数据库（迁移发生在打开内部）之前取得。账户 CLI 不使用本包。 +package instanceguard + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "strconv" + "strings" + + "agent-system-backend/internal/store" +) + +var ( + ErrBusy = errors.New("instance guard: sqlite service already running") + ErrHardlink = errors.New("instance guard: sqlite database has a hardlink alias") + ErrDBReplaced = errors.New("instance guard: sqlite database replaced") + ErrDBMissing = errors.New("instance guard: recorded sqlite database is missing") + ErrLockRebuilt = errors.New("instance guard: lock file was rebuilt") + ErrLockCorrupt = errors.New("instance guard: lock record is corrupt") + ErrInstanceMismatch = errors.New("instance guard: instance id mismatch") + ErrUnsupported = errors.New("instance guard: sqlite activity lock is not available on this platform") + ErrLockAliased = errors.New("instance guard: lock file aliases the sqlite database") +) + +// Record 是锁文件里记下的数据库身份。没有记录表示锁文件是新建的。 +type Record struct { + DBDev uint64 + DBIno uint64 + InstanceID string + LockGen string +} + +// Lease 持有活动服务锁，直到 Release。关闭顺序由调用方保证：先关存储，再 Release。 +type Lease struct { + fd int + dbPath string + lockDev uint64 + lockIno uint64 + dbDev uint64 + dbIno uint64 + dbExists bool + record *Record + parseErr error + release func() +} + +// Release 放开锁并关闭锁文件描述符。锁文件留在磁盘上。 +func (l *Lease) Release() { + if l == nil || l.release == nil { + return + } + l.release() + l.release = nil +} + +// DBPath 是解析后的数据库路径。符号链接指向真实文件。 +func (l *Lease) DBPath() string { + if l == nil { + return "" + } + return l.dbPath +} + +// AutomaticRefusal 在打开数据库之前调用。 +// 尚无记录时返回 nil（首次建立，含旧库升级）。硬链接在 Acquire 阶段已经拒绝。 +func (l *Lease) AutomaticRefusal() error { + if l.parseErr != nil { + return l.parseErr + } + if l.record == nil { + return nil + } + if !l.dbExists { + return ErrDBMissing + } + if l.record.DBDev != l.dbDev || l.record.DBIno != l.dbIno { + return ErrDBReplaced + } + return nil +} + +// BindOptions 控制守卫建立后的比对。 +type BindOptions struct { + Adopt bool + InstanceID string +} + +// Bind 在数据库已经打开并完成迁移之后调用。 +// 锁文件 inode 与表不一致则拒绝，除非 Adopt。通过时重写锁文件内容与表。 +func Bind(st *store.Store, lease *Lease, opt BindOptions) error { + if lease == nil { + return errors.New("instance guard: nil lease") + } + marker, ok, err := st.LoadInstanceGuard() + if err != nil { + return err + } + if ok && !opt.Adopt { + inodeMismatch := marker.LockDev != lease.lockDev || marker.LockIno != lease.lockIno + genMismatch := marker.LockGen != "" && (lease.record == nil || lease.record.LockGen != marker.LockGen) + if inodeMismatch || genMismatch { + return ErrLockRebuilt + } + } + id := opt.InstanceID + if ok && marker.InstanceID != "" && id != "" && marker.InstanceID != id && !opt.Adopt { + return ErrInstanceMismatch + } + if id == "" && ok { + id = marker.InstanceID + } + dev, ino, exists, err := lease.currentDB() + if err != nil { + return err + } + if !exists { + return ErrDBMissing + } + lease.dbDev, lease.dbIno, lease.dbExists = dev, ino, true + gen := "" + if ok && marker.LockGen != "" && !opt.Adopt { + gen = marker.LockGen + } else { + gen, err = newLockGen() + if err != nil { + return err + } + } + next := store.InstanceGuardMarker{ + DBDev: dev, DBIno: ino, LockDev: lease.lockDev, LockIno: lease.lockIno, + InstanceID: id, LockGen: gen, + } + if err := lease.writeRecord(dev, ino, id, gen); err != nil { + return err + } + if !ok || opt.Adopt || marker != next { + if err := st.SaveInstanceGuard(next); err != nil { + return err + } + } + return nil +} + +func newLockGen() (string, error) { + buf := make([]byte, 16) + if _, err := rand.Read(buf); err != nil { + return "", err + } + return hex.EncodeToString(buf), nil +} + +func parseRecord(raw string) (*Record, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, nil + } + parts := strings.Split(raw, "\t") + if parts[0] != "v1" { + return nil, ErrLockCorrupt + } + rec := &Record{} + seen := map[string]bool{} + for _, part := range parts[1:] { + key, val, ok := strings.Cut(part, "=") + if !ok || seen[key] || strings.ContainsAny(val, " \t") { + return nil, ErrLockCorrupt + } + seen[key] = true + switch key { + case "db_dev": + n, err := strconv.ParseUint(val, 10, 64) + if err != nil { + return nil, ErrLockCorrupt + } + rec.DBDev = n + case "db_ino": + n, err := strconv.ParseUint(val, 10, 64) + if err != nil { + return nil, ErrLockCorrupt + } + rec.DBIno = n + case "instance_id": + rec.InstanceID = val + case "lock_gen": + if val == "" { + return nil, ErrLockCorrupt + } + rec.LockGen = val + default: + return nil, ErrLockCorrupt + } + } + if !seen["db_dev"] || !seen["db_ino"] || !seen["instance_id"] { + return nil, ErrLockCorrupt + } + return rec, nil +} diff --git a/internal/instanceguard/guard_other.go b/internal/instanceguard/guard_other.go new file mode 100644 index 0000000..924f5ea --- /dev/null +++ b/internal/instanceguard/guard_other.go @@ -0,0 +1,18 @@ +//go:build !unix + +package instanceguard + +import "fmt" + +// Acquire 在非 unix 平台拒绝 SQLite 活动服务。PostgreSQL 档不调用本函数。 +func Acquire(string) (*Lease, error) { + return nil, ErrUnsupported +} + +func (l *Lease) writeRecord(uint64, uint64, string, string) error { + return ErrUnsupported +} + +func (l *Lease) currentDB() (uint64, uint64, bool, error) { + return 0, 0, false, fmt.Errorf("%w", ErrUnsupported) +} diff --git a/internal/instanceguard/guard_test.go b/internal/instanceguard/guard_test.go new file mode 100644 index 0000000..4a4dfb2 --- /dev/null +++ b/internal/instanceguard/guard_test.go @@ -0,0 +1,406 @@ +//go:build unix + +package instanceguard + +import ( + "errors" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "agent-system-backend/internal/store" +) + +func TestAcquireSameResourceAcrossAliases(t *testing.T) { + base := t.TempDir() + real := filepath.Join(base, "real") + if err := os.Mkdir(real, 0o755); err != nil { + t.Fatal(err) + } + link := filepath.Join(base, "link") + if err := os.Symlink(real, link); err != nil { + t.Fatal(err) + } + held, err := Acquire(link) + if err != nil { + t.Fatal(err) + } + defer held.Release() + if _, err := Acquire(real); !errors.Is(err, ErrBusy) { + t.Fatalf("symlink alias: %v", err) + } + + wd, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + if err := os.Chdir(base); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Chdir(wd) }) + if _, err := Acquire(filepath.Join(".", "real")); !errors.Is(err, ErrBusy) { + t.Fatalf("relative path: %v", err) + } +} + +func TestHardlinkRejected(t *testing.T) { + dir := t.TempDir() + db := filepath.Join(dir, "agent-system.db") + st, err := store.OpenSQLite(db) + if err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + if err := os.Link(db, filepath.Join(dir, "alias.db")); err != nil { + t.Fatal(err) + } + if _, err := Acquire(dir); !errors.Is(err, ErrHardlink) { + t.Fatalf("hardlink: %v", err) + } +} + +func TestDatabaseReplacementRefusedBeforeOpen(t *testing.T) { + dir := t.TempDir() + st, release, err := openAndBind(t, dir, "") + if err != nil { + t.Fatal(err) + } + db := filepath.Join(dir, "agent-system.db") + release() + _ = st + if err := os.Remove(db); err != nil { + t.Fatal(err) + } + raw, err := os.ReadFile(db + ".lock") + if err != nil { + t.Fatal(err) + } + rec, err := parseRecord(string(raw)) + if err != nil || rec == nil { + t.Fatal(err) + } + for i := 0; i < 16; i++ { + if err := os.WriteFile(db, []byte("replaced"), 0o600); err != nil { + t.Fatal(err) + } + _, ino, exists, err := statFile(db) + if err != nil { + t.Fatal(err) + } + if exists && ino != rec.DBIno { + break + } + if err := os.Remove(db); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "inode-burn-"+string(rune('a'+i))), []byte("x"), 0o600); err != nil { + t.Fatal(err) + } + } + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + defer lease.Release() + if err := lease.AutomaticRefusal(); !errors.Is(err, ErrDBReplaced) { + t.Fatalf("replaced db: %v", err) + } + if _, err := os.Stat(db); err != nil { + t.Fatal(err) + } +} + +func TestMissingRecordedDatabaseRefused(t *testing.T) { + dir := t.TempDir() + _, release, err := openAndBind(t, dir, "") + if err != nil { + t.Fatal(err) + } + release() + db := filepath.Join(dir, "agent-system.db") + if err := os.Remove(db); err != nil { + t.Fatal(err) + } + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + defer lease.Release() + if err := lease.AutomaticRefusal(); !errors.Is(err, ErrDBMissing) { + t.Fatalf("missing db: %v", err) + } +} + +func TestRebuiltLockRefusedUntilAdopt(t *testing.T) { + dir := t.TempDir() + st, release, err := openAndBind(t, dir, "inst-a") + if err != nil { + t.Fatal(err) + } + release() + _ = st.Close() + if err := os.Remove(filepath.Join(dir, "agent-system.db.lock")); err != nil { + t.Fatal(err) + } + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + if err := lease.AutomaticRefusal(); err != nil { + lease.Release() + t.Fatal(err) + } + opened, err := store.OpenSQLite(lease.DBPath()) + if err != nil { + lease.Release() + t.Fatal(err) + } + err = Bind(opened, lease, BindOptions{InstanceID: "inst-a"}) + _ = opened.Close() + lease.Release() + if !errors.Is(err, ErrLockRebuilt) { + t.Fatalf("rebuilt lock: %v", err) + } + + adopted, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + defer adopted.Release() + opened, err = store.OpenSQLite(adopted.DBPath()) + if err != nil { + t.Fatal(err) + } + defer opened.Close() + if err := Bind(opened, adopted, BindOptions{Adopt: true, InstanceID: "inst-a"}); err != nil { + t.Fatal(err) + } +} + +func TestClearedLockRecordRefusedWhenInodeStays(t *testing.T) { + dir := t.TempDir() + st, release, err := openAndBind(t, dir, "inst-a") + if err != nil { + t.Fatal(err) + } + release() + _ = st.Close() + lockPath := filepath.Join(dir, "agent-system.db.lock") + if err := os.Truncate(lockPath, 0); err != nil { + t.Fatal(err) + } + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + opened, err := store.OpenSQLite(lease.DBPath()) + if err != nil { + lease.Release() + t.Fatal(err) + } + err = Bind(opened, lease, BindOptions{InstanceID: "inst-a"}) + _ = opened.Close() + lease.Release() + if !errors.Is(err, ErrLockRebuilt) { + t.Fatalf("cleared lock: %v", err) + } +} + +func TestInstanceMismatchPreservesEmpty(t *testing.T) { + dir := t.TempDir() + st, release, err := openAndBind(t, dir, "inst-a") + if err != nil { + t.Fatal(err) + } + release() + _ = st.Close() + + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + opened, err := store.OpenSQLite(lease.DBPath()) + if err != nil { + lease.Release() + t.Fatal(err) + } + err = Bind(opened, lease, BindOptions{InstanceID: "inst-b"}) + _ = opened.Close() + lease.Release() + if !errors.Is(err, ErrInstanceMismatch) { + t.Fatalf("mismatch: %v", err) + } + + lease, err = Acquire(dir) + if err != nil { + t.Fatal(err) + } + defer lease.Release() + opened, err = store.OpenSQLite(lease.DBPath()) + if err != nil { + t.Fatal(err) + } + defer opened.Close() + if err := Bind(opened, lease, BindOptions{}); err != nil { + t.Fatal(err) + } + marker, ok, err := opened.LoadInstanceGuard() + if err != nil || !ok || marker.InstanceID != "inst-a" { + t.Fatalf("empty id should keep marker: ok=%v id=%q err=%v", ok, marker.InstanceID, err) + } +} + +func TestLockReleasedAfterKill(t *testing.T) { + if os.Getenv("U8_GUARD_ROLE") == "hold" { + lease, err := Acquire(os.Getenv("U8_GUARD_DIR")) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(os.Getenv("U8_GUARD_READY"), []byte("ready"), 0o600); err != nil { + t.Fatal(err) + } + time.Sleep(30 * time.Second) + lease.Release() + return + } + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + cmd := exec.Command(os.Args[0], "-test.run=TestLockReleasedAfterKill") + cmd.Env = append(os.Environ(), "U8_GUARD_ROLE=hold", "U8_GUARD_DIR="+dir, "U8_GUARD_READY="+ready) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + } + }) + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(ready); err == nil { + break + } + time.Sleep(20 * time.Millisecond) + } + if _, err := os.Stat(ready); err != nil { + t.Fatal(err) + } + if _, err := Acquire(dir); !errors.Is(err, ErrBusy) { + t.Fatalf("held: %v", err) + } + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + _, _ = cmd.Process.Wait() + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + lease.Release() +} + +func TestMissingDataDirIsCreatedBeforeLock(t *testing.T) { + dir := filepath.Join(t.TempDir(), "nested", "data") + lease, err := Acquire(dir) + if err != nil { + t.Fatal(err) + } + lease.Release() + if _, err := os.Stat(dir); err != nil { + t.Fatal(err) + } +} + +func TestLockSymlinkCannotAliasDatabase(t *testing.T) { + dir := t.TempDir() + db := filepath.Join(dir, "agent-system.db") + st, err := store.OpenSQLite(db) + if err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + if err := os.Remove(db + ".lock"); err != nil && !os.IsNotExist(err) { + t.Fatal(err) + } + if err := os.Symlink(db, db+".lock"); err != nil { + t.Fatal(err) + } + before, err := os.ReadFile(db) + if err != nil { + t.Fatal(err) + } + if _, err := Acquire(dir); !errors.Is(err, ErrLockAliased) { + t.Fatalf("symlink lock: %v", err) + } + after, err := os.ReadFile(db) + if err != nil { + t.Fatal(err) + } + if string(before) != string(after) { + t.Fatal("aliased lock wrote into the database") + } +} + +func TestSymlinkedDatabaseLocksResolvedFile(t *testing.T) { + base := t.TempDir() + realDir := filepath.Join(base, "real") + aliasDir := filepath.Join(base, "alias") + if err := os.Mkdir(realDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(aliasDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(realDir, "agent-system.db"), []byte{}, 0o600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(filepath.Join(realDir, "agent-system.db"), filepath.Join(aliasDir, "agent-system.db")); err != nil { + t.Fatal(err) + } + held, err := Acquire(aliasDir) + if err != nil { + t.Fatal(err) + } + defer held.Release() + if _, err := os.Stat(filepath.Join(realDir, "agent-system.db.lock")); err != nil { + t.Fatal(err) + } + if _, err := Acquire(realDir); !errors.Is(err, ErrBusy) { + t.Fatalf("resolved db lock: %v", err) + } +} + +func openAndBind(t *testing.T, dir, instanceID string) (*store.Store, func(), error) { + t.Helper() + lease, err := Acquire(dir) + if err != nil { + return nil, nil, err + } + if err := lease.AutomaticRefusal(); err != nil { + lease.Release() + return nil, nil, err + } + st, err := store.OpenSQLite(lease.DBPath()) + if err != nil { + lease.Release() + return nil, nil, err + } + if err := Bind(st, lease, BindOptions{InstanceID: instanceID}); err != nil { + _ = st.Close() + lease.Release() + return nil, nil, err + } + return st, func() { + _ = st.Close() + lease.Release() + }, nil +} diff --git a/internal/instanceguard/guard_unix.go b/internal/instanceguard/guard_unix.go new file mode 100644 index 0000000..ecc293f --- /dev/null +++ b/internal/instanceguard/guard_unix.go @@ -0,0 +1,201 @@ +//go:build unix + +package instanceguard + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "syscall" +) + +// Acquire 取得 SQLite 活动服务锁。失败时不返回 Lease。 +// 硬链接别名直接拒绝。锁文件被并发替换时重试数次后拒绝。 +func Acquire(dataDir string) (*Lease, error) { + dir, err := normalizeDataDir(dataDir) + if err != nil { + return nil, err + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + _, dbPath, err := resolveDB(dir) + if err != nil { + return nil, err + } + if err := rejectHardlink(dbPath); err != nil { + return nil, err + } + lockPath := dbPath + ".lock" + var last error + for attempt := 0; attempt < 5; attempt++ { + lease, retry, err := acquireOnce(dbPath, lockPath) + if err == nil && !retry { + return lease, nil + } + if retry { + last = err + continue + } + return nil, err + } + if last == nil { + last = errorsRetry() + } + return nil, last +} + +func errorsRetry() error { + return fmt.Errorf("instance guard: lock file inode changed while locking") +} + +func acquireOnce(dbPath, lockPath string) (*Lease, bool, error) { + fd, err := syscall.Open(lockPath, syscall.O_RDWR|syscall.O_CREAT|syscall.O_CLOEXEC|syscall.O_NOFOLLOW, 0o600) + if err != nil { + if err == syscall.ELOOP { + return nil, false, ErrLockAliased + } + return nil, false, err + } + fail := func(err error) (*Lease, bool, error) { + syscall.Close(fd) + return nil, false, err + } + if err := syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + syscall.Close(fd) + if err == syscall.EWOULDBLOCK || err == syscall.EAGAIN { + return nil, false, ErrBusy + } + return nil, false, err + } + var held syscall.Stat_t + if err := syscall.Fstat(fd, &held); err != nil { + return fail(err) + } + var named syscall.Stat_t + if err := syscall.Stat(lockPath, &named); err != nil { + return fail(err) + } + if held.Ino != named.Ino || held.Dev != named.Dev { + syscall.Flock(fd, syscall.LOCK_UN) + syscall.Close(fd) + return nil, true, errorsRetry() + } + buf := make([]byte, 512) + n, err := syscall.Read(fd, buf) + if err != nil { + return fail(err) + } + rec, perr := parseRecord(string(buf[:n])) + dbDev, dbIno, dbExists, err := statFile(dbPath) + if err != nil { + return fail(err) + } + if dbExists && dbDev == uint64(held.Dev) && dbIno == uint64(held.Ino) { + syscall.Flock(fd, syscall.LOCK_UN) + syscall.Close(fd) + return nil, false, ErrLockAliased + } + lease := &Lease{ + fd: fd, + dbPath: dbPath, + lockDev: uint64(held.Dev), + lockIno: uint64(held.Ino), + dbDev: dbDev, + dbIno: dbIno, + dbExists: dbExists, + record: rec, + parseErr: perr, + release: func() { + syscall.Flock(fd, syscall.LOCK_UN) + syscall.Close(fd) + }, + } + return lease, false, nil +} + +func (l *Lease) writeRecord(dev, ino uint64, instanceID, lockGen string) error { + if strings.ContainsAny(instanceID, " \t\r\n") { + return fmt.Errorf("instance guard: instance id contains whitespace") + } + line := fmt.Sprintf("v1\tdb_dev=%d\tdb_ino=%d\tinstance_id=%s\tlock_gen=%s\n", dev, ino, instanceID, lockGen) + if _, err := syscall.Seek(l.fd, 0, 0); err != nil { + return err + } + if _, err := syscall.Write(l.fd, []byte(line)); err != nil { + return err + } + return syscall.Ftruncate(l.fd, int64(len(line))) +} + +func (l *Lease) currentDB() (dev, ino uint64, exists bool, err error) { + return statFile(l.dbPath) +} + +func normalizeDataDir(dataDir string) (string, error) { + if strings.TrimSpace(dataDir) == "" { + dataDir = "." + } + abs, err := filepath.Abs(dataDir) + if err != nil { + return "", err + } + if _, err := os.Lstat(abs); err == nil { + return filepath.EvalSymlinks(abs) + } else if !os.IsNotExist(err) { + return "", err + } + parent, base := filepath.Split(abs) + parent = filepath.Clean(parent) + if _, err := os.Lstat(parent); err == nil { + resolved, err := filepath.EvalSymlinks(parent) + if err != nil { + return "", err + } + return filepath.Join(resolved, base), nil + } + return abs, nil +} + +func resolveDB(dataDir string) (logical, resolved string, err error) { + logical = filepath.Join(dataDir, "agent-system.db") + fi, err := os.Lstat(logical) + if err != nil { + if os.IsNotExist(err) { + return logical, logical, nil + } + return "", "", err + } + if fi.Mode()&os.ModeSymlink != 0 { + resolved, err = filepath.EvalSymlinks(logical) + return logical, resolved, err + } + return logical, logical, nil +} + +func rejectHardlink(dbPath string) error { + _, _, exists, err := statFile(dbPath) + if err != nil || !exists { + return err + } + var st syscall.Stat_t + if err := syscall.Stat(dbPath, &st); err != nil { + return err + } + if st.Nlink > 1 { + return ErrHardlink + } + return nil +} + +func statFile(path string) (dev, ino uint64, exists bool, err error) { + var st syscall.Stat_t + if err = syscall.Stat(path, &st); err != nil { + if os.IsNotExist(err) { + return 0, 0, false, nil + } + return 0, 0, false, err + } + return uint64(st.Dev), uint64(st.Ino), true, nil +} diff --git a/internal/store/durability.go b/internal/store/durability.go new file mode 100644 index 0000000..b295b3c --- /dev/null +++ b/internal/store/durability.go @@ -0,0 +1,67 @@ +package store + +import ( + "fmt" + "strings" +) + +// DurabilityRefusal 判断 PostgreSQL 耐久参数。off 或缺失即拒绝。 +// synchronous_commit=local 不是 off，不拒绝。 +func DurabilityRefusal(settings map[string]string) error { + for _, key := range []string{"fsync", "full_page_writes", "synchronous_commit"} { + value, ok := settings[key] + if !ok || strings.EqualFold(strings.TrimSpace(value), "off") { + return fmt.Errorf("durability: %s=%q", key, value) + } + } + return nil +} + +func (s *Store) verifyDurability() error { + switch s.dialect.name() { + case "sqlite": + return s.verifySQLiteDurability() + case "postgres": + return s.verifyPostgresDurability() + default: + return fmt.Errorf("durability: unknown dialect %s", s.dialect.name()) + } +} + +func (s *Store) verifySQLiteDurability() error { + var journal string + if err := s.db.QueryRow(`PRAGMA journal_mode`).Scan(&journal); err != nil { + return fmt.Errorf("durability: journal_mode: %w", err) + } + var syncLevel int64 + if err := s.db.QueryRow(`PRAGMA synchronous`).Scan(&syncLevel); err != nil { + return fmt.Errorf("durability: synchronous: %w", err) + } + if !strings.EqualFold(journal, "wal") || syncLevel != 2 { + return fmt.Errorf("durability: journal_mode=%s synchronous=%d", journal, syncLevel) + } + return nil +} + +func (s *Store) verifyPostgresDurability() error { + rows, err := s.db.Query(`SELECT name, setting FROM pg_settings WHERE name IN ('fsync','full_page_writes','synchronous_commit')`) + if err != nil { + return fmt.Errorf("durability: pg_settings: %w", err) + } + defer rows.Close() + got := map[string]string{} + for rows.Next() { + var name, setting string + if err := rows.Scan(&name, &setting); err != nil { + return err + } + got[name] = setting + } + if err := rows.Err(); err != nil { + return err + } + if err := DurabilityRefusal(got); err != nil { + return err + } + return nil +} diff --git a/internal/store/durability_test.go b/internal/store/durability_test.go new file mode 100644 index 0000000..5f7beac --- /dev/null +++ b/internal/store/durability_test.go @@ -0,0 +1,95 @@ +package store + +import ( + "os" + "path/filepath" + "strings" + "testing" + + core "agent-system-core" +) + +func TestDurabilityRefusal(t *testing.T) { + ok := map[string]string{ + "fsync": "on", "full_page_writes": "on", "synchronous_commit": "local", + } + if err := DurabilityRefusal(ok); err != nil { + t.Fatal(err) + } + for _, key := range []string{"fsync", "full_page_writes", "synchronous_commit"} { + bad := map[string]string{ + "fsync": "on", "full_page_writes": "on", "synchronous_commit": "on", + } + bad[key] = "off" + if err := DurabilityRefusal(bad); err == nil { + t.Fatalf("%s off should refuse", key) + } + delete(bad, key) + if err := DurabilityRefusal(bad); err == nil { + t.Fatalf("missing %s should refuse", key) + } + } +} + +func TestSQLitePragmaReadback(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "agent-system.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + var journal string + if err := st.db.QueryRow(`PRAGMA journal_mode`).Scan(&journal); err != nil { + t.Fatal(err) + } + var syncLevel int64 + if err := st.db.QueryRow(`PRAGMA synchronous`).Scan(&syncLevel); err != nil { + t.Fatal(err) + } + if !strings.EqualFold(journal, "wal") || syncLevel != 2 { + t.Fatalf("journal=%s synchronous=%d", journal, syncLevel) + } + marker, ok, err := st.LoadInstanceGuard() + if err != nil || ok || marker.InstanceID != "" { + t.Fatalf("fresh db has no guard row: ok=%v err=%v", ok, err) + } +} + +func TestOpenExistingDoesNotCreateMissingParent(t *testing.T) { + missing := filepath.Join(t.TempDir(), "absent", "agent-system.db") + if _, err := OpenExistingSQLite(missing); err == nil { + t.Fatal("missing file opened") + } + if _, err := os.Stat(filepath.Dir(missing)); !os.IsNotExist(err) { + t.Fatalf("parent created: %v", err) + } +} + +func TestRunSurvivesReopen(t *testing.T) { + path := filepath.Join(t.TempDir(), "agent-system.db") + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + run := &core.Run{ + ID: "run_keep", ConversationID: "c", ProjectID: "p", Owner: "o", + Kind: core.RunConversation, Status: core.RunWaitingHITL, + } + if err := st.CreateRun(run); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + st, err = OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + defer st.Close() + got, err := st.GetRunByID("run_keep") + if err != nil { + t.Fatal(err) + } + if got.Status != core.RunWaitingHITL { + t.Fatalf("status %s", got.Status) + } +} diff --git a/internal/store/durability_unix_test.go b/internal/store/durability_unix_test.go new file mode 100644 index 0000000..14c4f87 --- /dev/null +++ b/internal/store/durability_unix_test.go @@ -0,0 +1,35 @@ +//go:build unix + +package store + +import ( + "os/signal" + "path/filepath" + "strings" + "syscall" + "testing" +) + +func TestWriteFailureIsReturned(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "agent-system.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + signal.Ignore(syscall.SIGXFSZ) + t.Cleanup(func() { signal.Reset(syscall.SIGXFSZ) }) + var lim syscall.Rlimit + if err := syscall.Getrlimit(syscall.RLIMIT_FSIZE, &lim); err != nil { + t.Fatal(err) + } + old := lim + lim.Cur = 4096 + if err := syscall.Setrlimit(syscall.RLIMIT_FSIZE, &lim); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = syscall.Setrlimit(syscall.RLIMIT_FSIZE, &old) }) + _, err = st.CreateProject("owner", strings.Repeat("x", 64*1024)) + if err == nil { + t.Fatal("oversized write returned success") + } +} diff --git a/internal/store/instance_guard.go b/internal/store/instance_guard.go new file mode 100644 index 0000000..7265ef1 --- /dev/null +++ b/internal/store/instance_guard.go @@ -0,0 +1,47 @@ +package store + +import ( + "database/sql" + "errors" +) + +// InstanceGuardMarker 是 instance_guard 表的唯一一行。 +type InstanceGuardMarker struct { + DBDev uint64 + DBIno uint64 + LockDev uint64 + LockIno uint64 + InstanceID string + LockGen string +} + +// LoadInstanceGuard 读取守卫标记。没有行时 ok 为 false。 +func (s *Store) LoadInstanceGuard() (InstanceGuardMarker, bool, error) { + var m InstanceGuardMarker + var dbDev, dbIno, lockDev, lockIno int64 + err := s.db.QueryRow(`SELECT db_dev, db_ino, lock_dev, lock_ino, instance_id, lock_gen FROM instance_guard WHERE id=1`). + Scan(&dbDev, &dbIno, &lockDev, &lockIno, &m.InstanceID, &m.LockGen) + if errors.Is(err, sql.ErrNoRows) { + return InstanceGuardMarker{}, false, nil + } + if err != nil { + return InstanceGuardMarker{}, false, err + } + m.DBDev = uint64(dbDev) + m.DBIno = uint64(dbIno) + m.LockDev = uint64(lockDev) + m.LockIno = uint64(lockIno) + return m, true, nil +} + +// SaveInstanceGuard 写入或覆盖唯一一行。 +func (s *Store) SaveInstanceGuard(m InstanceGuardMarker) error { + _, err := s.db.Exec(s.q(`INSERT INTO instance_guard(id, db_dev, db_ino, lock_dev, lock_ino, instance_id, lock_gen) + VALUES(1,?,?,?,?,?,?) + ON CONFLICT(id) DO UPDATE SET + db_dev=excluded.db_dev, db_ino=excluded.db_ino, + lock_dev=excluded.lock_dev, lock_ino=excluded.lock_ino, + instance_id=excluded.instance_id, lock_gen=excluded.lock_gen`), + int64(m.DBDev), int64(m.DBIno), int64(m.LockDev), int64(m.LockIno), m.InstanceID, m.LockGen) + return err +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index a175ae4..42867e1 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -901,6 +901,34 @@ func init() { up: map[string][]string{"sqlite": execEnvUp, "postgres": execEnvUp}, down: map[string][]string{"sqlite": execEnvDown, "postgres": execEnvDown}, }, + // U8：活动服务锁的库内标记。锁文件 inode 与本行不一致表示锁文件被重建。 + migration{version: 36, name: "instance_guard", + up: map[string][]string{"sqlite": { + `CREATE TABLE IF NOT EXISTS instance_guard ( + id INTEGER PRIMARY KEY CHECK (id = 1), + db_dev INTEGER NOT NULL, + db_ino INTEGER NOT NULL, + lock_dev INTEGER NOT NULL, + lock_ino INTEGER NOT NULL, + instance_id TEXT NOT NULL DEFAULT '', + lock_gen TEXT NOT NULL DEFAULT '' + )`, + }, "postgres": { + `CREATE TABLE IF NOT EXISTS instance_guard ( + id INTEGER PRIMARY KEY CHECK (id = 1), + db_dev BIGINT NOT NULL, + db_ino BIGINT NOT NULL, + lock_dev BIGINT NOT NULL, + lock_ino BIGINT NOT NULL, + instance_id TEXT NOT NULL DEFAULT '', + lock_gen TEXT NOT NULL DEFAULT '' + )`, + }}, + down: map[string][]string{"sqlite": { + `DROP TABLE IF EXISTS instance_guard`, + }, "postgres": { + `DROP TABLE IF EXISTS instance_guard`, + }}}, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index fd96457..c3ddeee 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -40,6 +40,7 @@ var migrationBreaksOldReaders = map[int]bool{ 33: true, // 本人证明与 grant 绑定协议；旧恢复写者和兑换者必须排除 34: false, // 游标签名密钥只新增内部列；旧读者按列名读取既有字段 35: false, // exec_env 新表；旧读者不读这些表 + 36: false, // instance_guard 是新表，旧制品不读取它 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/store.go b/internal/store/store.go index 5d30b74..da1cc6a 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -189,7 +189,8 @@ func openSQLiteFile(path string, createArtifacts, migrateSchema bool) (*Store, e // 跨进程迁移/写路径依赖 busy_timeout 排队而不是中途 SQLITE_BUSY（KTD8）。 // busy_timeout 30s：迁移竞争测试在慢速 CI 盘上 4 个迁移者串行全程可超 5s // （实测 job 2285 SQLITE_BUSY）；本地档宁可排队不报错。 - dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_txlock=immediate", path) + // synchronous(FULL) 是恢复方案 KTD14 的可靠档。打开后在同一连接读回，见 verifyDurability。 + dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)&_pragma=synchronous(FULL)&_pragma=foreign_keys(1)&_txlock=immediate", path) db, err := sql.Open("sqlite", dsn) if err != nil { return nil, err @@ -293,6 +294,10 @@ func finishOpenMode(db *sql.DB, d dialect, artifactsDir string, blobs blobstore. db.Close() return nil, err } + if err := s.verifyDurability(); err != nil { + db.Close() + return nil, err + } if createArtifacts { if err := os.MkdirAll(artifactsDir, 0o755); err != nil { db.Close() From 3187a009f684f91d933dbb39304be44a90a2a05e Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 18 Oct 2026 09:29:02 +0000 Subject: [PATCH 5/9] Record the EXT-20 regulations provider task card Why: Track U9 base contract delivery for the regulations KB plan. Evidence: none (task record). Impact: Task record only. Refs: Agent-System-Dev .task/feature-regulations-kb.md Co-authored-by: Cursor --- .task/feature-regulations-kb-ext20.md | 29 +++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 .task/feature-regulations-kb-ext20.md diff --git a/.task/feature-regulations-kb-ext20.md b/.task/feature-regulations-kb-ext20.md new file mode 100644 index 0000000..eaa5383 --- /dev/null +++ b/.task/feature-regulations-kb-ext20.md @@ -0,0 +1,29 @@ +# feature/regulations-kb-ext20 任务卡 + +任务主记录：本文件随 PR 描述维护。 +状态：进行中；owner：weichen。 + +## 基线与授权 + +- 权威仓：ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-System-Backend.git；root_target：main；immediate_target：main +- branch_kind：standalone；跨仓总任务：Agent-System-Dev `feature/regulations-kb`（[Dev PR #614](https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Dev/pulls/614)） +- R0（initial_main_sha）：cac60c0；工作区：/home/weichen/projects/Agent-System-Backend-wt-feature-regulations-kb-ext20；分支：feature/regulations-kb-ext20 +- 授权与交付终点：用户 2026-10-18 /goal 指示按法规知识库计划实现全部内容；本任务交付 U9 的基座公开契约 EXT-20（提案：Documents `docs/proposals/2026-10-18-ext-20-regulations-provider.md`，分支 `feature/regulations-kb-u9-contract`），经 PR 合入 main。不含法规提供方实现（属 KB 仓）与 RDM 消费（U6）。 +- 实施执行：agent-per-unit（配置：AGENT-EXECUTION.md） + +## 目标与非目标 + +- 目标：`pkg/extension` 新增 `RegulationsProvider`/`RegulationsRegistrar`/`RegulationsHost` 与 DTO；进程内注册槽仅接受插件 `kbreg` 且只注册一次；分发 Host 实现四个方法并强制调用方身份、查询最小化（机密词拒绝、长度限制）、状态透传与无明文审计；快照、边界与契约/兼容测试。 +- 非目标：不实现搜索或语料检索；不改 `KnowledgeSearch` 语义。 + +## Agent 实施收据 + +| 单元 ID | 预期范围 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 | +|---|---|---|---|---|---|---|---|---|---| +| 待定义 | — | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — | + +## 结果与验证（完成后填写） + +- 结果： +- 验证： +- 影响与剩余项： From aaed108d7903ddeed314d644ba3fdd1a176e685e Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Sun, 18 Oct 2026 09:38:04 +0000 Subject: [PATCH 6/9] agent-exec transport cfc2e055f212789112cf11716f5649c5/c04b36e76696 --- internal/app/app.go | 2 +- internal/app/compiled.go | 19 ++ internal/app/hostcaps.go | 1 + internal/app/regulations.go | 220 +++++++++++++++ internal/app/regulations_compat_test.go | 134 +++++++++ internal/app/regulations_test.go | 347 ++++++++++++++++++++++++ pkg/extension/extension.go | 4 +- pkg/extension/regulations.go | 209 ++++++++++++++ pkg/extension/regulations_test.go | 78 ++++++ pkg/extension/surface.txt | 46 ++++ 10 files changed, 1058 insertions(+), 2 deletions(-) create mode 100644 internal/app/regulations.go create mode 100644 internal/app/regulations_compat_test.go create mode 100644 internal/app/regulations_test.go create mode 100644 pkg/extension/regulations.go create mode 100644 pkg/extension/regulations_test.go diff --git a/internal/app/app.go b/internal/app/app.go index a6edf18..1b1ec7e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -649,7 +649,7 @@ func Run(opt Options) error { if err != nil { return err } - caps := hostcaps{projects: projectSlot, directory: directorySlot, st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc, allowImplicitLocal: !authOn, + caps := hostcaps{projects: projectSlot, directory: directorySlot, regulations: ®ulationsSlot{}, st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc, allowImplicitLocal: !authOn, accountGuard: func(owner string) error { valid, err := effectiveAccount(context.Background(), owner) if err != nil { diff --git a/internal/app/compiled.go b/internal/app/compiled.go index 644978a..6ae2d7a 100644 --- a/internal/app/compiled.go +++ b/internal/app/compiled.go @@ -127,6 +127,17 @@ func (r pluginRegistry) RegisterAccountDirectoryScopeResolver(fn extension.Accou return r.caps.directory.register(r.pluginID, fn) } +func (r compiledRegistry) RegisterRegulationsProvider(extension.RegulationsProvider) error { + return errors.New("extension: declarative plugins cannot register a regulations provider") +} + +func (r pluginRegistry) RegisterRegulationsProvider(p extension.RegulationsProvider) error { + if r.caps.regulations == nil { + return errors.New("extension: regulations provider registration unavailable") + } + return r.caps.regulations.register(r.pluginID, p) +} + // RegisterResourceAuthorizer 进程内扩展（EXT-1）：填充 EXT-8 适配器槽 // （hostcaps.authz 共享指针；首个注册者生效，组合根已装配进 Server/runmgr）。 // 此前 pluginRegistry 未覆盖该方法——进程内插件注册适配器会命中声明式分支的 @@ -169,6 +180,14 @@ func (r recordRegistry) RegisterAccountDirectoryScopeResolver(fn extension.Accou return registrar.RegisterAccountDirectoryScopeResolver(fn) } +func (r recordRegistry) RegisterRegulationsProvider(p extension.RegulationsProvider) error { + registrar, ok := r.Registry.(extension.RegulationsRegistrar) + if !ok { + return errors.New("extension: regulations provider registration unavailable") + } + return registrar.RegisterRegulationsProvider(p) +} + func (r recordRegistry) AddCommand(spec json.RawMessage) error { if err := r.Registry.AddCommand(spec); err != nil { return err diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go index 7ffd00c..101a4a2 100644 --- a/internal/app/hostcaps.go +++ b/internal/app/hostcaps.go @@ -23,6 +23,7 @@ import ( type hostcaps struct { projects *projectLifecycleSlot directory *accountDirectorySlot + regulations *regulationsSlot // EXT-20 法规提供方槽（共享指针；nil/未注册＝不可用） accountGuard func(string) error allowImplicitLocal bool st *store.Store diff --git a/internal/app/regulations.go b/internal/app/regulations.go new file mode 100644 index 0000000..f8e2061 --- /dev/null +++ b/internal/app/regulations.go @@ -0,0 +1,220 @@ +package app + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "strings" + "sync" + "unicode/utf8" + + "agent-system-backend/pkg/extension" +) + +// EXT-20：共享法规检索与网络证据。提供方由受信插件 kbreg 进程内注册；基座只 +// 填写调用方身份、做查询最小化校验并审计，不改写提供方返回的状态。 + +const ( + regulationsPluginID = "kbreg" + maxWebQueryRunes = 256 + + auditRegulationsSearch = "regulations.search" + auditRegulationsEvidence = "regulations.evidence" + auditWebSearch = "web.search" + auditWebCapture = "web.capture" + + regulationsAuditInvalid = "invalid" + regulationsAuditError = "error" + regulationsAuditOK = "ok" +) + +var ( + errRegulationsSchema = errors.New("extension: unsupported regulations schema version") + errWebQueryInvalid = errors.New("extension: web search query must be 1-256 characters") +) + +type regulationsSlot struct { + mu sync.RWMutex + provider extension.RegulationsProvider +} + +func (s *regulationsSlot) register(pluginID string, p extension.RegulationsProvider) error { + if p == nil || pluginID != regulationsPluginID { + return errors.New("extension: regulations provider requires trusted kbreg registration") + } + s.mu.Lock() + defer s.mu.Unlock() + if s.provider != nil { + return errors.New("extension: regulations provider already registered") + } + s.provider = p + return nil +} + +func (s *regulationsSlot) get() extension.RegulationsProvider { + if s == nil { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + return s.provider +} + +// regulationsAudit is the only content written to the audit trail: never the +// query text, excerpts or captured page text. +type regulationsAudit struct { + Revision string `json:"revision,omitempty"` + Count int `json:"count"` + State string `json:"state"` + QuerySHA256 string `json:"querySha256,omitempty"` +} + +func regulationsQueryDigest(query string) string { + if query == "" { + return "" + } + sum := sha256.Sum256([]byte(query)) + return hex.EncodeToString(sum[:]) +} + +// regulations resolves the provider and the trusted caller. Without an audit +// store the capability is unavailable: calls must never run unaudited. +func (h inprocHost) regulations() (extension.RegulationsProvider, extension.RegulationsCaller, error) { + if err := h.requireEffective(); err != nil { + return nil, extension.RegulationsCaller{}, err + } + p := h.caps.regulations.get() + if p == nil || h.caps.st == nil { + return nil, extension.RegulationsCaller{}, extension.ErrRegulationsUnavailable + } + return p, extension.RegulationsCaller{PluginID: h.pluginID, Actor: h.actor, Owner: h.owner}, nil +} + +// auditRegulations appends one event per call. A call error takes precedence; +// otherwise an audit failure withholds the result. +func (h inprocHost) auditRegulations(action string, rec regulationsAudit, callErr error) error { + if callErr != nil && rec.State == "" { + rec.State = regulationsAuditError + } + evidence, err := json.Marshal(rec) + if err == nil { + err = h.Audit(action, regulationsPluginID, string(evidence)) + } + if callErr != nil { + return callErr + } + return err +} + +func (h inprocHost) SearchRegulations(ctx context.Context, req extension.RegulationsSearchRequest) (extension.RegulationsSearchResult, error) { + p, caller, err := h.regulations() + if err != nil { + return extension.RegulationsSearchResult{}, err + } + rec := regulationsAudit{Revision: req.CorpusRevision, QuerySHA256: regulationsQueryDigest(req.Query)} + if req.SchemaVersion != extension.RegulationsSchemaVersion { + rec.State = regulationsAuditInvalid + return extension.RegulationsSearchResult{}, h.auditRegulations(auditRegulationsSearch, rec, errRegulationsSchema) + } + res, err := p.SearchRegulations(ctx, caller, req) + if err == nil { + rec.Revision, rec.Count, rec.State = res.CorpusRevision, len(res.Hits), res.State + } + if err := h.auditRegulations(auditRegulationsSearch, rec, err); err != nil { + return extension.RegulationsSearchResult{}, err + } + return res, nil +} + +func (h inprocHost) ResolveRegulationEvidence(ctx context.Context, req extension.RegulationEvidenceRequest) (extension.RegulationEvidence, error) { + p, caller, err := h.regulations() + if err != nil { + return extension.RegulationEvidence{}, err + } + rec := regulationsAudit{Revision: req.CorpusRevision} + if req.SchemaVersion != extension.RegulationsSchemaVersion { + rec.State = regulationsAuditInvalid + return extension.RegulationEvidence{}, h.auditRegulations(auditRegulationsEvidence, rec, errRegulationsSchema) + } + ev, err := p.ResolveRegulationEvidence(ctx, caller, req) + if err == nil { + rec.Count, rec.State = 1, regulationsAuditOK + if ev.CorpusRevision != "" { + rec.Revision = ev.CorpusRevision + } + } + if err := h.auditRegulations(auditRegulationsEvidence, rec, err); err != nil { + return extension.RegulationEvidence{}, err + } + return ev, nil +} + +// webQueryExcluded reports whether query contains any non-empty exclusion +// term, case-insensitively. +func webQueryExcluded(query string, exclusions []string) bool { + lower := strings.ToLower(query) + for _, term := range exclusions { + term = strings.ToLower(strings.TrimSpace(term)) + if term != "" && strings.Contains(lower, term) { + return true + } + } + return false +} + +func (h inprocHost) SearchWeb(ctx context.Context, req extension.WebSearchRequest) (extension.WebSearchResult, error) { + p, caller, err := h.regulations() + if err != nil { + return extension.WebSearchResult{}, err + } + req.Query = strings.TrimSpace(req.Query) + rec := regulationsAudit{QuerySHA256: regulationsQueryDigest(req.Query)} + if req.SchemaVersion != extension.RegulationsSchemaVersion { + rec.State = regulationsAuditInvalid + return extension.WebSearchResult{}, h.auditRegulations(auditWebSearch, rec, errRegulationsSchema) + } + if req.Query == "" || !utf8.ValidString(req.Query) || utf8.RuneCountInString(req.Query) > maxWebQueryRunes { + rec.State = regulationsAuditInvalid + return extension.WebSearchResult{}, h.auditRegulations(auditWebSearch, rec, errWebQueryInvalid) + } + if webQueryExcluded(req.Query, req.Exclusions) { + res := extension.WebSearchResult{State: extension.WebSearchStateRejected, Reason: "query contains an excluded term"} + rec.State = res.State + if err := h.auditRegulations(auditWebSearch, rec, nil); err != nil { + return extension.WebSearchResult{}, err + } + return res, nil + } + res, err := p.SearchWeb(ctx, caller, req) + if err == nil { + rec.Count, rec.State = len(res.Results), res.State + } + if err := h.auditRegulations(auditWebSearch, rec, err); err != nil { + return extension.WebSearchResult{}, err + } + return res, nil +} + +func (h inprocHost) CaptureWeb(ctx context.Context, req extension.WebCaptureRequest) (extension.WebCapture, error) { + p, caller, err := h.regulations() + if err != nil { + return extension.WebCapture{}, err + } + rec := regulationsAudit{QuerySHA256: regulationsQueryDigest(strings.TrimSpace(req.SearchQuery))} + if req.SchemaVersion != extension.RegulationsSchemaVersion { + rec.State = regulationsAuditInvalid + return extension.WebCapture{}, h.auditRegulations(auditWebCapture, rec, errRegulationsSchema) + } + capture, err := p.CaptureWeb(ctx, caller, req) + if err == nil { + rec.Count, rec.State = len(capture.Candidates), capture.State + } + if err := h.auditRegulations(auditWebCapture, rec, err); err != nil { + return extension.WebCapture{}, err + } + return capture, nil +} + +var _ extension.RegulationsHost = inprocHost{} diff --git a/internal/app/regulations_compat_test.go b/internal/app/regulations_compat_test.go new file mode 100644 index 0000000..e91fc18 --- /dev/null +++ b/internal/app/regulations_compat_test.go @@ -0,0 +1,134 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "testing" + "time" + + agentcommand "agent-command" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" +) + +// compatRegulationsPlugin stands in for the external kbreg repository: it uses +// only pkg/extension types and registers through the optional Registrar. +type compatRegulationsPlugin struct{ seen []extension.RegulationsCaller } + +func (p *compatRegulationsPlugin) Register(reg extension.Registry) error { + registrar, ok := reg.(extension.RegulationsRegistrar) + if !ok { + return errors.New("regulations registrar unavailable") + } + return registrar.RegisterRegulationsProvider(p) +} + +func (p *compatRegulationsPlugin) SearchRegulations(_ context.Context, c extension.RegulationsCaller, req extension.RegulationsSearchRequest) (extension.RegulationsSearchResult, error) { + p.seen = append(p.seen, c) + return extension.RegulationsSearchResult{CorpusRevision: "rev-compat", State: extension.RegulationsStateOK, + Hits: []extension.RegulationHit{{RegulationID: "r", EditionID: "e", NodeID: "n", Disclosure: extension.RegulationDisclosureExcerpt}}}, nil +} + +func (p *compatRegulationsPlugin) ResolveRegulationEvidence(_ context.Context, c extension.RegulationsCaller, req extension.RegulationEvidenceRequest) (extension.RegulationEvidence, error) { + p.seen = append(p.seen, c) + return extension.RegulationEvidence{CorpusRevision: req.CorpusRevision, NodeID: req.NodeID, Disclosure: extension.RegulationDisclosureFull, HashVerified: true}, nil +} + +func (p *compatRegulationsPlugin) SearchWeb(_ context.Context, c extension.RegulationsCaller, req extension.WebSearchRequest) (extension.WebSearchResult, error) { + p.seen = append(p.seen, c) + return extension.WebSearchResult{Provider: "compat", SentQuery: req.Query, State: extension.WebSearchStateEmpty}, nil +} + +func (p *compatRegulationsPlugin) CaptureWeb(_ context.Context, c extension.RegulationsCaller, req extension.WebCaptureRequest) (extension.WebCapture, error) { + p.seen = append(p.seen, c) + return extension.WebCapture{FinalURL: req.URL, State: extension.WebCaptureStateFailed, Reason: "offline"}, nil +} + +// compatRegulationsConsumer is a second compiled plugin that reaches the +// provider only through the dispatch Host type assertion. +func compatRegulationsConsumer(reg extension.Registry) error { + return reg.AddCommandHandler(json.RawMessage(`{"id":"consumer.regs","title":"regs"}`), + func(ctx context.Context, h extension.Host, req extension.Request) (any, error) { + regs, ok := h.(extension.RegulationsHost) + if !ok { + return nil, errors.New("regulations host capability missing") + } + search, err := regs.SearchRegulations(ctx, extension.RegulationsSearchRequest{SchemaVersion: extension.RegulationsSchemaVersion, Query: "注册检验"}) + if err != nil { + return nil, err + } + web, err := regs.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: extension.RegulationsSchemaVersion, Query: "注册检验 指导原则"}) + if err != nil { + return nil, err + } + return map[string]any{"revision": search.CorpusRevision, "webState": web.State}, nil + }) +} + +func newRegulationsCompatApp(t *testing.T, plugins ...extension.InProc) (*cmdbus.Bus, *store.Store, hostcaps) { + t.Helper() + extension.ResetCompiled() + t.Cleanup(extension.ResetCompiled) + for _, p := range plugins { + extension.RegisterCompiled(p) + } + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "compat.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + if err := st.EnsureUser("alice", ""); err != nil { + t.Fatal(err) + } + bus := cmdbus.New(st, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() }) + caps := hostcaps{st: st, regulations: ®ulationsSlot{}} + applyCompiled("", bus, caps) + return bus, st, caps +} + +func TestRegulationsCompiledPluginCompatibility(t *testing.T) { + provider := &compatRegulationsPlugin{} + impostor := &compatRegulationsPlugin{} + bus, st, caps := newRegulationsCompatApp(t, + extension.InProc{ID: "impostor", Register: impostor.Register}, + extension.InProc{ID: "kbreg", Register: provider.Register}, + extension.InProc{ID: "consumer", Register: compatRegulationsConsumer}, + ) + if caps.regulations.get() != extension.RegulationsProvider(provider) { + t.Fatal("compiled kbreg provider was not the registered provider") + } + resp := bus.Dispatch(context.Background(), cmdbus.Request{ID: "consumer.regs", Owner: "alice", Actor: "alice", Entrance: "button"}) + if resp.Receipt.State != agentcommand.StateCommitted { + t.Fatalf("dispatch = %+v", resp.Receipt) + } + payload, _ := resp.Payload.(map[string]any) + if payload["revision"] != "rev-compat" || payload["webState"] != extension.WebSearchStateEmpty { + t.Fatalf("payload = %+v", resp.Payload) + } + want := extension.RegulationsCaller{PluginID: "consumer", Actor: "alice", Owner: "alice"} + if len(provider.seen) != 2 || provider.seen[0] != want || provider.seen[1] != want || len(impostor.seen) != 0 { + t.Fatalf("provider callers = %+v impostor = %+v", provider.seen, impostor.seen) + } + evs, err := st.ListAuditEvents("alice", 20) + if err != nil { + t.Fatal(err) + } + actions := map[string]bool{} + for _, ev := range evs { + actions[ev.Action] = true + } + if !actions[auditRegulationsSearch] || !actions[auditWebSearch] { + t.Fatalf("host audit missing: %+v", evs) + } +} + +func TestRegulationsCompiledConsumerWithoutProvider(t *testing.T) { + bus, _, _ := newRegulationsCompatApp(t, extension.InProc{ID: "consumer", Register: compatRegulationsConsumer}) + resp := bus.Dispatch(context.Background(), cmdbus.Request{ID: "consumer.regs", Owner: "alice", Actor: "alice", Entrance: "button"}) + if resp.Receipt.State == agentcommand.StateCommitted { + t.Fatalf("consumer succeeded without a regulations provider: %+v", resp.Payload) + } +} diff --git a/internal/app/regulations_test.go b/internal/app/regulations_test.go new file mode 100644 index 0000000..7d279eb --- /dev/null +++ b/internal/app/regulations_test.go @@ -0,0 +1,347 @@ +package app + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" +) + +type fakeRegulationsProvider struct { + callers []extension.RegulationsCaller + webQueries []string + search extension.RegulationsSearchResult + evidence extension.RegulationEvidence + web extension.WebSearchResult + capture extension.WebCapture + err error +} + +func (f *fakeRegulationsProvider) SearchRegulations(_ context.Context, c extension.RegulationsCaller, _ extension.RegulationsSearchRequest) (extension.RegulationsSearchResult, error) { + f.callers = append(f.callers, c) + return f.search, f.err +} + +func (f *fakeRegulationsProvider) ResolveRegulationEvidence(_ context.Context, c extension.RegulationsCaller, _ extension.RegulationEvidenceRequest) (extension.RegulationEvidence, error) { + f.callers = append(f.callers, c) + return f.evidence, f.err +} + +func (f *fakeRegulationsProvider) SearchWeb(_ context.Context, c extension.RegulationsCaller, req extension.WebSearchRequest) (extension.WebSearchResult, error) { + f.callers = append(f.callers, c) + f.webQueries = append(f.webQueries, req.Query) + return f.web, f.err +} + +func (f *fakeRegulationsProvider) CaptureWeb(_ context.Context, c extension.RegulationsCaller, _ extension.WebCaptureRequest) (extension.WebCapture, error) { + f.callers = append(f.callers, c) + return f.capture, f.err +} + +func newRegulationsTestHost(t *testing.T) (inprocHost, *store.Store) { + t.Helper() + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "regulations.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + if err := st.EnsureUser("alice", ""); err != nil { + t.Fatal(err) + } + caps := hostcaps{st: st, regulations: ®ulationsSlot{}} + return inprocHost{caps: caps, pluginID: "rdm", owner: "alice", actor: "alice"}, st +} + +func registerFakeRegulations(t *testing.T, h inprocHost) *fakeRegulationsProvider { + t.Helper() + p := &fakeRegulationsProvider{} + if err := (pluginRegistry{pluginID: "kbreg", caps: h.caps}).RegisterRegulationsProvider(p); err != nil { + t.Fatal(err) + } + return p +} + +const regV = extension.RegulationsSchemaVersion + +func callAllRegulations(h inprocHost, version int) []error { + ctx := context.Background() + _, e1 := h.SearchRegulations(ctx, extension.RegulationsSearchRequest{SchemaVersion: version, Query: "医疗器械 临床评价"}) + _, e2 := h.ResolveRegulationEvidence(ctx, extension.RegulationEvidenceRequest{SchemaVersion: version, CorpusRevision: "rev-1", NodeID: "n1"}) + _, e3 := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: version, Query: "临床评价 指导原则"}) + _, e4 := h.CaptureWeb(ctx, extension.WebCaptureRequest{SchemaVersion: version, URL: "https://www.nmpa.gov.cn/x", SearchQuery: "临床评价 指导原则"}) + return []error{e1, e2, e3, e4} +} + +func TestRegulationsUnavailableWithoutProvider(t *testing.T) { + h, st := newRegulationsTestHost(t) + for i, err := range callAllRegulations(h, regV) { + if !errors.Is(err, extension.ErrRegulationsUnavailable) { + t.Fatalf("method %d without provider: %v", i, err) + } + } + h.caps.regulations = nil + for i, err := range callAllRegulations(h, regV) { + if !errors.Is(err, extension.ErrRegulationsUnavailable) { + t.Fatalf("method %d without slot: %v", i, err) + } + } + if evs, err := st.ListAuditEvents("alice", 10); err != nil || len(evs) != 0 { + t.Fatalf("unavailable calls audited: %+v %v", evs, err) + } +} + +func TestRegulationsRegistrationOnlyTrustedKbregOnce(t *testing.T) { + h, _ := newRegulationsTestHost(t) + first, second := &fakeRegulationsProvider{}, &fakeRegulationsProvider{} + if err := (pluginRegistry{pluginID: "rdm", caps: h.caps}).RegisterRegulationsProvider(first); err == nil { + t.Fatal("non-kbreg plugin registered a regulations provider") + } + kbreg := pluginRegistry{pluginID: "kbreg", caps: h.caps} + if err := kbreg.RegisterRegulationsProvider(nil); err == nil { + t.Fatal("nil provider registered") + } + if err := kbreg.RegisterRegulationsProvider(first); err != nil { + t.Fatal(err) + } + if err := kbreg.RegisterRegulationsProvider(second); err == nil { + t.Fatal("duplicate registration accepted") + } + if h.caps.regulations.get() != first { + t.Fatal("duplicate registration replaced the first provider") + } + if err := (pluginRegistry{pluginID: "kbreg"}).RegisterRegulationsProvider(first); err == nil { + t.Fatal("registration without a host slot accepted") + } +} + +func TestRegulationsDeclarativeRegistrationRejected(t *testing.T) { + h, _ := newRegulationsTestHost(t) + p := &fakeRegulationsProvider{} + if err := (compiledRegistry{}).RegisterRegulationsProvider(p); err == nil { + t.Fatal("declarative registry accepted a regulations provider") + } + if err := (recordRegistry{Registry: pluginRegistry{pluginID: "kbreg", caps: h.caps}}).RegisterRegulationsProvider(p); err != nil { + t.Fatalf("record registry did not forward kbreg registration: %v", err) + } + if h.caps.regulations.get() != p { + t.Fatal("forwarded registration did not fill the slot") + } +} + +func TestRegulationsCallerFilledFromDispatchContext(t *testing.T) { + h, _ := newRegulationsTestHost(t) + h.actor = "alice-session" + p := registerFakeRegulations(t, h) + p.web.State = extension.WebSearchStateOK + for i, err := range callAllRegulations(h, regV) { + if err != nil { + t.Fatalf("method %d: %v", i, err) + } + } + want := extension.RegulationsCaller{PluginID: "rdm", Actor: "alice-session", Owner: "alice"} + if len(p.callers) != 4 { + t.Fatalf("provider calls = %d", len(p.callers)) + } + for i, got := range p.callers { + if got != want { + t.Fatalf("call %d caller = %+v, want %+v", i, got, want) + } + } +} + +func TestRegulationsRejectsOtherSchemaVersions(t *testing.T) { + h, _ := newRegulationsTestHost(t) + p := registerFakeRegulations(t, h) + for _, version := range []int{0, regV + 1} { + for i, err := range callAllRegulations(h, version) { + if !errors.Is(err, errRegulationsSchema) { + t.Fatalf("schema %d method %d: %v", version, i, err) + } + } + } + if len(p.callers) != 0 { + t.Fatalf("provider reached with unsupported schema: %d calls", len(p.callers)) + } +} + +func TestRegulationsWebQueryBounds(t *testing.T) { + h, _ := newRegulationsTestHost(t) + p := registerFakeRegulations(t, h) + ctx := context.Background() + for _, q := range []string{"", " \t\n", strings.Repeat("法", maxWebQueryRunes+1)} { + if _, err := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: q}); !errors.Is(err, errWebQueryInvalid) { + t.Fatalf("query of %d runes accepted: %v", len([]rune(q)), err) + } + } + if len(p.callers) != 0 { + t.Fatal("provider reached with an invalid query") + } + maxQuery := strings.Repeat("法", maxWebQueryRunes) + if _, err := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: " " + maxQuery + " "}); err != nil { + t.Fatalf("256-rune query rejected: %v", err) + } + if len(p.webQueries) != 1 || p.webQueries[0] != maxQuery { + t.Fatalf("provider query was not trimmed: %q", p.webQueries) + } +} + +func TestRegulationsExclusionRejectedBeforeProvider(t *testing.T) { + h, st := newRegulationsTestHost(t) + p := registerFakeRegulations(t, h) + ctx := context.Background() + query := "Project-Falcon 临床评价 指导原则" + res, err := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: query, + Exclusions: []string{"", " ", "project-falcon"}}) + if err != nil || res.State != extension.WebSearchStateRejected || res.SentQuery != "" || len(res.Results) != 0 { + t.Fatalf("excluded query result = %+v %v", res, err) + } + if len(p.callers) != 0 { + t.Fatal("provider received a query containing an excluded term") + } + evs, err := st.ListAuditEvents("alice", 10) + if err != nil || len(evs) != 1 || evs[0].Action != auditWebSearch { + t.Fatalf("rejection audit: %+v %v", evs, err) + } + if strings.Contains(strings.ToLower(evs[0].Evidence), "falcon") || !strings.Contains(evs[0].Evidence, `"state":"rejected"`) { + t.Fatalf("rejection audit evidence = %s", evs[0].Evidence) + } + if _, err := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: query, Exclusions: []string{"", " "}}); err != nil { + t.Fatal(err) + } + if len(p.callers) != 1 { + t.Fatal("blank exclusions blocked an allowed query") + } +} + +func TestRegulationsProviderStatesPassThroughUnchanged(t *testing.T) { + h, _ := newRegulationsTestHost(t) + p := registerFakeRegulations(t, h) + at := time.Date(2026, 10, 18, 9, 0, 0, 0, time.UTC) + p.search = extension.RegulationsSearchResult{CorpusRevision: "rev-7", State: extension.RegulationsStateDegraded, + DegradedReasons: []string{"vector_index_unavailable"}, + Hits: []extension.RegulationHit{{RegulationID: "r1", EditionID: "e1", NodeID: "n1", + Disclosure: extension.RegulationDisclosureWithheld, WithheldReason: "rights"}, + {RegulationID: "r2", EditionID: "e2", NodeID: "n2", Disclosure: extension.RegulationDisclosureMetadataOnly}}} + p.evidence = extension.RegulationEvidence{CorpusRevision: "rev-7", NodeID: "n1", + Disclosure: extension.RegulationDisclosureWithheld, WithheldReason: "rights", HashVerified: false} + p.web = extension.WebSearchResult{Provider: "fake", SentQuery: "q", SearchedAt: at, + State: extension.WebSearchStateRateLimited, Reason: "quota"} + p.capture = extension.WebCapture{CaptureID: "c1", FetchedAt: at, Retention: extension.WebRetentionMetadataOnly, + State: extension.WebCaptureStateRejected, Reason: "robots"} + ctx := context.Background() + search, err := h.SearchRegulations(ctx, extension.RegulationsSearchRequest{SchemaVersion: regV, Query: "q"}) + if err != nil || !reflect.DeepEqual(search, p.search) { + t.Fatalf("search = %+v %v", search, err) + } + evidence, err := h.ResolveRegulationEvidence(ctx, extension.RegulationEvidenceRequest{SchemaVersion: regV, CorpusRevision: "rev-7", NodeID: "n1"}) + if err != nil || !reflect.DeepEqual(evidence, p.evidence) { + t.Fatalf("evidence = %+v %v", evidence, err) + } + web, err := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: "q"}) + if err != nil || !reflect.DeepEqual(web, p.web) { + t.Fatalf("web = %+v %v", web, err) + } + p.web.State = extension.WebSearchStateRejected + web, err = h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: "q"}) + if err != nil || web.State != extension.WebSearchStateRejected { + t.Fatalf("provider rejection rewritten: %+v %v", web, err) + } + capture, err := h.CaptureWeb(ctx, extension.WebCaptureRequest{SchemaVersion: regV, URL: "https://example.org"}) + if err != nil || !reflect.DeepEqual(capture, p.capture) { + t.Fatalf("capture = %+v %v", capture, err) + } + p.err = errors.New("provider failure") + if _, err := h.SearchRegulations(ctx, extension.RegulationsSearchRequest{SchemaVersion: regV, Query: "q"}); !errors.Is(err, p.err) { + t.Fatalf("provider failure wrapped as success: %v", err) + } +} + +func TestRegulationsAuditEvidenceOmitsQueryAndText(t *testing.T) { + h, st := newRegulationsTestHost(t) + p := registerFakeRegulations(t, h) + const ( + secretQuery = "第十七条 secret-query-text" + secretExcerpt = "SECRET-EXCERPT 原文片段" + secretPage = "SECRET-PAGE 网页全文" + ) + p.search = extension.RegulationsSearchResult{CorpusRevision: "rev-9", State: extension.RegulationsStateOK, + Hits: []extension.RegulationHit{{NodeID: "n1", Excerpt: secretExcerpt, Disclosure: extension.RegulationDisclosureExcerpt}}} + p.evidence = extension.RegulationEvidence{CorpusRevision: "rev-9", NodeID: "n1", Excerpt: secretExcerpt, + Disclosure: extension.RegulationDisclosureFull, HashVerified: true} + p.web = extension.WebSearchResult{SentQuery: secretQuery, State: extension.WebSearchStateOK, + Results: []extension.WebSearchHit{{Rank: 1, URL: "https://example.org", Snippet: secretExcerpt}}} + p.capture = extension.WebCapture{State: extension.WebCaptureStateCaptured, Retention: extension.WebRetentionFull, + Text: secretPage, Candidates: []extension.WebClauseCandidate{{Excerpt: secretExcerpt}}} + ctx := context.Background() + if _, err := h.SearchRegulations(ctx, extension.RegulationsSearchRequest{SchemaVersion: regV, Query: secretQuery}); err != nil { + t.Fatal(err) + } + if _, err := h.ResolveRegulationEvidence(ctx, extension.RegulationEvidenceRequest{SchemaVersion: regV, CorpusRevision: "rev-9", NodeID: "n1"}); err != nil { + t.Fatal(err) + } + if _, err := h.SearchWeb(ctx, extension.WebSearchRequest{SchemaVersion: regV, Query: " " + secretQuery + " "}); err != nil { + t.Fatal(err) + } + if _, err := h.CaptureWeb(ctx, extension.WebCaptureRequest{SchemaVersion: regV, URL: "https://example.org", SearchQuery: secretQuery}); err != nil { + t.Fatal(err) + } + evs, err := st.ListAuditEvents("alice", 10) + if err != nil || len(evs) != 4 { + t.Fatalf("audit events = %+v %v", evs, err) + } + sum := sha256.Sum256([]byte(secretQuery)) + digest := hex.EncodeToString(sum[:]) + want := map[string]regulationsAudit{ + auditRegulationsSearch: {Revision: "rev-9", Count: 1, State: "ok", QuerySHA256: digest}, + auditRegulationsEvidence: {Revision: "rev-9", Count: 1, State: "ok"}, + auditWebSearch: {Count: 1, State: "ok", QuerySHA256: digest}, + auditWebCapture: {Count: 1, State: "captured", QuerySHA256: digest}, + } + for _, ev := range evs { + for _, secret := range []string{"secret-query-text", "第十七条", "SECRET-EXCERPT", "原文片段", "SECRET-PAGE", "网页全文"} { + if strings.Contains(ev.Evidence, secret) || strings.Contains(ev.Target, secret) { + t.Fatalf("%s audit leaks %q: %s", ev.Action, secret, ev.Evidence) + } + } + var fields map[string]any + if err := json.Unmarshal([]byte(ev.Evidence), &fields); err != nil { + t.Fatal(err) + } + for k := range fields { + if k != "revision" && k != "count" && k != "state" && k != "querySha256" { + t.Fatalf("%s audit has unexpected field %q", ev.Action, k) + } + } + var got regulationsAudit + if err := json.Unmarshal([]byte(ev.Evidence), &got); err != nil { + t.Fatal(err) + } + if w, ok := want[ev.Action]; !ok || got != w || ev.Owner != "alice" { + t.Fatalf("%s audit = %+v owner=%s, want %+v", ev.Action, got, ev.Owner, w) + } + delete(want, ev.Action) + } +} + +func TestRegulationsRequireEffectiveAccount(t *testing.T) { + h, _ := newRegulationsTestHost(t) + p := registerFakeRegulations(t, h) + h.caps.accountGuard = func(string) error { return store.ErrAccountForbidden } + for i, err := range callAllRegulations(h, regV) { + if !errors.Is(err, store.ErrAccountForbidden) { + t.Fatalf("method %d ignored the account guard: %v", i, err) + } + } + if len(p.callers) != 0 { + t.Fatal("provider reached for an ineffective account") + } +} diff --git a/pkg/extension/extension.go b/pkg/extension/extension.go index d43a6ea..1d0e1fc 100644 --- a/pkg/extension/extension.go +++ b/pkg/extension/extension.go @@ -40,7 +40,9 @@ import ( // AccountDirectoryRegistrar for scoped, current account candidate decisions. // 2.14.0-experimental：InProc.HostOptions / CLIHook，以及 MergeHostOptions、 // RegisterCLIFlags、RunCLIHooks（组合二进制把插件声明并进基座启动，不覆盖显式参数）。 -const SurfaceVersion = "2.14.0-experimental" +// 2.15.0-experimental：EXT-20 可选 RegulationsHost 与受信进程内 RegulationsRegistrar +// （共享法规检索与网络证据；仅插件 kbreg 可注册提供方）。 +const SurfaceVersion = "2.15.0-experimental" // ErrNotLinked 表示进程没有空白导入 link 包，Main 无法启动服务。 var ErrNotLinked = errors.New("extension: runner not linked; blank-import agent-system-backend/pkg/extension/link") diff --git a/pkg/extension/regulations.go b/pkg/extension/regulations.go new file mode 100644 index 0000000..21bf58b --- /dev/null +++ b/pkg/extension/regulations.go @@ -0,0 +1,209 @@ +package extension + +import ( + "context" + "errors" + "time" +) + +// RegulationsSchemaVersion is the only request schema the host accepts for +// EXT-20; any other value is rejected before reaching the provider. +const RegulationsSchemaVersion = 1 + +const ( + RegulationsStateOK = "ok" + RegulationsStateEmpty = "empty" + RegulationsStateDegraded = "degraded" + RegulationsStateUnavailable = "unavailable" + + RegulationDisclosureFull = "full" + RegulationDisclosureExcerpt = "excerpt" + RegulationDisclosureMetadataOnly = "metadata_only" + RegulationDisclosureWithheld = "withheld" + + WebSearchStateOK = "ok" + WebSearchStateEmpty = "empty" + WebSearchStateUnavailable = "unavailable" + WebSearchStateRateLimited = "rate_limited" + WebSearchStateRejected = "rejected" + WebSearchStateError = "error" + + WebCaptureStateCaptured = "captured" + WebCaptureStateRejected = "rejected" + WebCaptureStateFailed = "failed" + + WebRetentionFull = "full" + WebRetentionExcerpt = "excerpt" + WebRetentionMetadataOnly = "metadata_only" +) + +// ErrRegulationsUnavailable is returned by every RegulationsHost method when no +// trusted provider is registered. Callers must not fall back to KnowledgeSearch. +var ErrRegulationsUnavailable = errors.New("extension: regulations provider unavailable") + +// RegulationsCaller is filled by the host from the trusted dispatch context. +// It deliberately has no project ID: shared regulation access neither widens +// nor narrows with a project. +type RegulationsCaller struct { + PluginID string `json:"pluginId"` + Actor string `json:"actor,omitempty"` + Owner string `json:"owner"` +} + +// RegulationsSearchRequest searches the controlled regulation corpus. An empty +// CorpusRevision means the current published revision; the result echoes the +// revision actually searched. +type RegulationsSearchRequest struct { + SchemaVersion int `json:"schemaVersion"` + Query string `json:"query"` + CorpusRevision string `json:"corpusRevision,omitempty"` + Kinds []string `json:"kinds,omitempty"` + CurrentOnly bool `json:"currentOnly,omitempty"` + DocNo string `json:"docNo,omitempty"` + ClauseNo string `json:"clauseNo,omitempty"` + Limit int `json:"limit,omitempty"` +} + +type RegulationsSearchResult struct { + CorpusRevision string `json:"corpusRevision"` + State string `json:"state"` // ok|empty|degraded|unavailable + DegradedReasons []string `json:"degradedReasons,omitempty"` + Hits []RegulationHit `json:"hits"` +} + +type RegulationHit struct { + RegulationID string `json:"regulationId"` + EditionID string `json:"editionId"` + NodeID string `json:"nodeId"` + DocNo string `json:"docNo,omitempty"` + Title string `json:"title,omitempty"` + EditionLabel string `json:"editionLabel,omitempty"` + LegalStatus string `json:"legalStatus,omitempty"` + ClauseNo string `json:"clauseNo,omitempty"` + HeadingPath []string `json:"headingPath,omitempty"` + Excerpt string `json:"excerpt,omitempty"` + ExcerptStart int `json:"excerptStart,omitempty"` + ExcerptEnd int `json:"excerptEnd,omitempty"` + NodeSHA256 string `json:"nodeSha256,omitempty"` + SourceAssetSHA256 string `json:"sourceAssetSha256,omitempty"` + PageStart int `json:"pageStart,omitempty"` + PageEnd int `json:"pageEnd,omitempty"` + Disclosure string `json:"disclosure"` // full|excerpt|metadata_only|withheld + WithheldReason string `json:"withheldReason,omitempty"` + IsSummary bool `json:"isSummary,omitempty"` + MatchKinds []string `json:"matchKinds,omitempty"` + Score float64 `json:"score,omitempty"` +} + +// RegulationEvidenceRequest retrieves the original text span of a node pinned +// to a corpus revision. +type RegulationEvidenceRequest struct { + SchemaVersion int `json:"schemaVersion"` + CorpusRevision string `json:"corpusRevision"` + NodeID string `json:"nodeId"` + Start int `json:"start,omitempty"` + End int `json:"end,omitempty"` +} + +type RegulationEvidence struct { + CorpusRevision string `json:"corpusRevision"` + RegulationID string `json:"regulationId,omitempty"` + EditionID string `json:"editionId,omitempty"` + NodeID string `json:"nodeId"` + ClauseNo string `json:"clauseNo,omitempty"` + Excerpt string `json:"excerpt,omitempty"` + Start int `json:"start,omitempty"` + End int `json:"end,omitempty"` + NodeSHA256 string `json:"nodeSha256,omitempty"` + SourceAssetSHA256 string `json:"sourceAssetSha256,omitempty"` + PageStart int `json:"pageStart,omitempty"` + PageEnd int `json:"pageEnd,omitempty"` + Disclosure string `json:"disclosure"` + WithheldReason string `json:"withheldReason,omitempty"` + HashVerified bool `json:"hashVerified"` +} + +// WebSearchRequest performs a real web search. Exclusions are caller-declared +// confidential terms (project names, numbers, people); a query containing any +// of them is rejected by the host before the provider sees it. +type WebSearchRequest struct { + SchemaVersion int `json:"schemaVersion"` + Query string `json:"query"` + Exclusions []string `json:"exclusions,omitempty"` + PreferOfficial bool `json:"preferOfficial,omitempty"` + Limit int `json:"limit,omitempty"` +} + +type WebSearchResult struct { + Provider string `json:"provider,omitempty"` + SentQuery string `json:"sentQuery,omitempty"` + SearchedAt time.Time `json:"searchedAt"` + State string `json:"state"` // ok|empty|unavailable|rate_limited|rejected|error + Reason string `json:"reason,omitempty"` + Results []WebSearchHit `json:"results,omitempty"` +} + +type WebSearchHit struct { + Rank int `json:"rank"` + URL string `json:"url"` + Title string `json:"title,omitempty"` + Snippet string `json:"snippet,omitempty"` + Publisher string `json:"publisher,omitempty"` + Official bool `json:"official,omitempty"` +} + +// WebCaptureRequest fetches a page found by a prior SearchWeb call. +type WebCaptureRequest struct { + SchemaVersion int `json:"schemaVersion"` + URL string `json:"url"` + SearchProvider string `json:"searchProvider,omitempty"` + SearchQuery string `json:"searchQuery,omitempty"` + SearchRank int `json:"searchRank,omitempty"` +} + +type WebCapture struct { + CaptureID string `json:"captureId,omitempty"` + FinalURL string `json:"finalUrl,omitempty"` + FetchedAt time.Time `json:"fetchedAt"` + ContentType string `json:"contentType,omitempty"` + ByteSize int64 `json:"byteSize,omitempty"` + SHA256 string `json:"sha256,omitempty"` + Retention string `json:"retention,omitempty"` // full|excerpt|metadata_only + // Text is present only when the source rights allow full retention. + Text string `json:"text,omitempty"` + Candidates []WebClauseCandidate `json:"candidates,omitempty"` + State string `json:"state"` // captured|rejected|failed + Reason string `json:"reason,omitempty"` +} + +type WebClauseCandidate struct { + ClauseNo string `json:"clauseNo,omitempty"` + Excerpt string `json:"excerpt"` + Start int `json:"start"` + End int `json:"end"` +} + +// RegulationsProvider is implemented and registered by the trusted regulations +// knowledge plugin (plugin ID "kbreg"); the host ships no implementation. +type RegulationsProvider interface { + SearchRegulations(context.Context, RegulationsCaller, RegulationsSearchRequest) (RegulationsSearchResult, error) + ResolveRegulationEvidence(context.Context, RegulationsCaller, RegulationEvidenceRequest) (RegulationEvidence, error) + SearchWeb(context.Context, RegulationsCaller, WebSearchRequest) (WebSearchResult, error) + CaptureWeb(context.Context, RegulationsCaller, WebCaptureRequest) (WebCapture, error) +} + +// RegulationsRegistrar is optional on a compiled in-process Registry. Only +// plugin ID "kbreg" may register, and only once. +type RegulationsRegistrar interface { + RegisterRegulationsProvider(RegulationsProvider) error +} + +// RegulationsHost is optional on dispatch Host handles (obtained by type +// assertion). A missing capability means unavailable; callers must not fall +// back to KnowledgeSearch. +type RegulationsHost interface { + SearchRegulations(context.Context, RegulationsSearchRequest) (RegulationsSearchResult, error) + ResolveRegulationEvidence(context.Context, RegulationEvidenceRequest) (RegulationEvidence, error) + SearchWeb(context.Context, WebSearchRequest) (WebSearchResult, error) + CaptureWeb(context.Context, WebCaptureRequest) (WebCapture, error) +} diff --git a/pkg/extension/regulations_test.go b/pkg/extension/regulations_test.go new file mode 100644 index 0000000..f0bd49a --- /dev/null +++ b/pkg/extension/regulations_test.go @@ -0,0 +1,78 @@ +package extension_test + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "agent-system-backend/pkg/extension" +) + +// publicRegulationsProvider is a downstream provider built only from the +// public extension surface. +type publicRegulationsProvider struct{} + +func (publicRegulationsProvider) SearchRegulations(context.Context, extension.RegulationsCaller, extension.RegulationsSearchRequest) (extension.RegulationsSearchResult, error) { + return extension.RegulationsSearchResult{State: extension.RegulationsStateEmpty}, nil +} + +func (publicRegulationsProvider) ResolveRegulationEvidence(context.Context, extension.RegulationsCaller, extension.RegulationEvidenceRequest) (extension.RegulationEvidence, error) { + return extension.RegulationEvidence{Disclosure: extension.RegulationDisclosureMetadataOnly}, nil +} + +func (publicRegulationsProvider) SearchWeb(context.Context, extension.RegulationsCaller, extension.WebSearchRequest) (extension.WebSearchResult, error) { + return extension.WebSearchResult{State: extension.WebSearchStateUnavailable}, nil +} + +func (publicRegulationsProvider) CaptureWeb(context.Context, extension.RegulationsCaller, extension.WebCaptureRequest) (extension.WebCapture, error) { + return extension.WebCapture{State: extension.WebCaptureStateFailed}, nil +} + +var _ extension.RegulationsProvider = publicRegulationsProvider{} + +func TestRegulationsContractIsOptional(t *testing.T) { + var host extension.Host + if _, ok := host.(extension.RegulationsHost); ok { + t.Fatal("nil host must not expose regulations") + } + var registry extension.Registry + if _, ok := registry.(extension.RegulationsRegistrar); ok { + t.Fatal("nil registry must not expose regulations registration") + } + if extension.RegulationsSchemaVersion != 1 { + t.Fatalf("schema version = %d", extension.RegulationsSchemaVersion) + } + if extension.ErrRegulationsUnavailable == nil { + t.Fatal("unavailable error must be defined") + } +} + +func TestRegulationsWireNames(t *testing.T) { + states := []string{extension.RegulationsStateOK, extension.RegulationsStateEmpty, extension.RegulationsStateDegraded, + extension.RegulationsStateUnavailable, extension.RegulationDisclosureFull, extension.RegulationDisclosureExcerpt, + extension.RegulationDisclosureMetadataOnly, extension.RegulationDisclosureWithheld, extension.WebSearchStateOK, + extension.WebSearchStateEmpty, extension.WebSearchStateUnavailable, extension.WebSearchStateRateLimited, + extension.WebSearchStateRejected, extension.WebSearchStateError, extension.WebCaptureStateCaptured, + extension.WebCaptureStateRejected, extension.WebCaptureStateFailed, extension.WebRetentionFull, + extension.WebRetentionExcerpt, extension.WebRetentionMetadataOnly} + want := "ok empty degraded unavailable full excerpt metadata_only withheld ok empty unavailable rate_limited " + + "rejected error captured rejected failed full excerpt metadata_only" + if got := strings.Join(states, " "); got != want { + t.Fatalf("wire states = %q", got) + } + raw, err := json.Marshal(extension.WebSearchRequest{SchemaVersion: extension.RegulationsSchemaVersion, Query: "q"}) + if err != nil { + t.Fatal(err) + } + if string(raw) != `{"schemaVersion":1,"query":"q"}` { + t.Fatalf("optional fields not omitted: %s", raw) + } + raw, err = json.Marshal(extension.RegulationsCaller{PluginID: "rdm", Owner: "alice"}) + if err != nil { + t.Fatal(err) + } + if strings.Contains(strings.ToLower(string(raw)), "project") { + t.Fatalf("caller must not carry a project: %s", raw) + } +} diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index 3fe18e7..be390a3 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -30,6 +30,7 @@ Compiled DelegatedProjectProjectionHost DelegatedProjectProjectionHost.ProjectProjectionForCreator ErrNotLinked +ErrRegulationsUnavailable Handler Host Host.ArtifactContent @@ -78,6 +79,33 @@ Registry.AddSkill Registry.Note Registry.RegisterResourceAuthorizer Registry.RegisterResourceAuthorizerV2 +RegulationDisclosureExcerpt +RegulationDisclosureFull +RegulationDisclosureMetadataOnly +RegulationDisclosureWithheld +RegulationEvidence +RegulationEvidenceRequest +RegulationHit +RegulationsCaller +RegulationsHost +RegulationsHost.CaptureWeb +RegulationsHost.ResolveRegulationEvidence +RegulationsHost.SearchRegulations +RegulationsHost.SearchWeb +RegulationsProvider +RegulationsProvider.CaptureWeb +RegulationsProvider.ResolveRegulationEvidence +RegulationsProvider.SearchRegulations +RegulationsProvider.SearchWeb +RegulationsRegistrar +RegulationsRegistrar.RegisterRegulationsProvider +RegulationsSchemaVersion +RegulationsSearchRequest +RegulationsSearchResult +RegulationsStateDegraded +RegulationsStateEmpty +RegulationsStateOK +RegulationsStateUnavailable Request ResetCompiled ResourceAuthorizer @@ -93,3 +121,21 @@ SurfaceHost.SurfaceList SurfaceHost.SurfaceSave SurfaceSaveRequest SurfaceVersion +WebCapture +WebCaptureRequest +WebCaptureStateCaptured +WebCaptureStateFailed +WebCaptureStateRejected +WebClauseCandidate +WebRetentionExcerpt +WebRetentionFull +WebRetentionMetadataOnly +WebSearchHit +WebSearchRequest +WebSearchResult +WebSearchStateEmpty +WebSearchStateError +WebSearchStateOK +WebSearchStateRateLimited +WebSearchStateRejected +WebSearchStateUnavailable From 276dd73824116fb55064eefef17f069ee1545ada Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Sun, 18 Oct 2026 09:44:31 +0000 Subject: [PATCH 7/9] agent-exec transport 473a7db9497fd0ab028ed664d4e26aec/87e91bc00054 --- deps/development-manager | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deps/development-manager b/deps/development-manager index bd3a7de..67784f4 160000 --- a/deps/development-manager +++ b/deps/development-manager @@ -1 +1 @@ -Subproject commit bd3a7de57a3a9e1a4495b94c5fa5c08d71e7be8c +Subproject commit 67784f43eb0430bf596c2cda685cbb4946d2e032 From 379336604810ebd8a7477ffaacdda7cd8f51f1ed Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 18 Oct 2026 09:51:50 +0000 Subject: [PATCH 8/9] Record the EXT-20 implementation receipt and verification Why: Track U9 delivery evidence. Evidence: Targeted vet/test PASS at aaed108; full suite only fails a pre-existing environment test. Impact: Card only. Refs: Agent-System-Dev feature/regulations-kb Co-authored-by: Cursor --- .task/feature-regulations-kb-ext20.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/.task/feature-regulations-kb-ext20.md b/.task/feature-regulations-kb-ext20.md index eaa5383..df89105 100644 --- a/.task/feature-regulations-kb-ext20.md +++ b/.task/feature-regulations-kb-ext20.md @@ -20,10 +20,11 @@ | 单元 ID | 预期范围 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 | |---|---|---|---|---|---|---|---|---|---| -| 待定义 | — | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — | +| backend-ext20-regulations-provider | pkg/extension、internal/app | implementation | cursor, antigravity, pi | 1 | cursor-agent-stable / auto（session not-observable，新进程） | success | refs/agent-exec/cfc2e055f212789112cf11716f5649c5/c04b36e76696 → aaed108（apply 快进） | 否 | /tmp/regkb-u9-receipt.json | ## 结果与验证（完成后填写） -- 结果： -- 验证： -- 影响与剩余项： +- 结果：`pkg/extension/regulations.go`（EXT-20 接口、DTO、状态常量、`ErrRegulationsUnavailable`）、`surface.txt` 与 `SurfaceVersion` 2.15.0-experimental；`internal/app/regulations.go` 注册槽（仅 kbreg、一次、非 nil；声明式拒绝）与 inprocHost 四方法（有效账户检查、调用方取自分发上下文、schema 校验、查询去空白/1–256 字符/排除词拒绝且不调用提供方、状态透传、审计仅含修订/计数/状态/查询 sha256）；单元与兼容测试。 +- 验证（SELF，host 于 HEAD aaed108）：`go vet ./pkg/extension/ ./internal/app/` 与 `go test ./pkg/extension/... ./internal/app/...` PASS；全量 `GOWORK=off go vet ./... && go test -count=1 ./...` 40 包 ok，仅 `pkg/identity TestDSNFileMode` 失败——在未改动基线 cac60c0 上同样失败（本机环境问题，本变更未触及 pkg/identity），以 CI（GATE）为准。 +- Review：host 主模型审阅实际 diff（base 3187a00 / head aaed108），与提案第 2、4、5 节一致，无阻塞项。 +- 影响与剩余项：新增可选能力，未注册提供方时返回不可用，既有行为不变。 From 499ad471530ccbfe81b8c8e5a9456334f55a897d Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 18 Oct 2026 09:52:33 +0000 Subject: [PATCH 9/9] Close the EXT-20 task card before merge Why: Card closes in the same PR per delivery rules. Evidence: Card only. Impact: None. Refs: https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Backend/pulls/212 Co-authored-by: Cursor --- .task/feature-regulations-kb-ext20.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.task/feature-regulations-kb-ext20.md b/.task/feature-regulations-kb-ext20.md index df89105..7f2f349 100644 --- a/.task/feature-regulations-kb-ext20.md +++ b/.task/feature-regulations-kb-ext20.md @@ -1,7 +1,7 @@ # feature/regulations-kb-ext20 任务卡 任务主记录：本文件随 PR 描述维护。 -状态：进行中；owner：weichen。 +状态：已完成（合入前闭合）；owner：weichen。PR：https://git-home.u1680832.nyat.app:30319/weichen/Agent-System-Backend/pulls/212 ## 基线与授权