Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0c7961839f | ||
|
|
dc47c10d68 |
@@ -31,6 +31,8 @@ func main() {
|
||||
os.Exit(runConfig(os.Args[2:]))
|
||||
case "user":
|
||||
os.Exit(runUser(os.Args[2:]))
|
||||
case "identity":
|
||||
os.Exit(runIdentity(os.Args[2:]))
|
||||
}
|
||||
}
|
||||
fs := flag.NewFlagSet("serve", flag.ExitOnError)
|
||||
@@ -283,3 +285,50 @@ func runUser(args []string) int {
|
||||
return 2
|
||||
}
|
||||
}
|
||||
|
||||
// runIdentity:U1 I3 服务实例库内身份(manifest.instance_id ↔ agent_system_identity)。
|
||||
// 与 serve --instance(runs 进程标识)无关。
|
||||
func runIdentity(args []string) int {
|
||||
if len(args) == 0 {
|
||||
fmt.Fprintln(os.Stderr, "用法:agent-system identity <ensure|verify> --instance-id ID [--dbref REF] [--data DIR] [--dsn DSN]")
|
||||
return 2
|
||||
}
|
||||
sub := args[0]
|
||||
fs := flag.NewFlagSet("identity "+sub, flag.ExitOnError)
|
||||
data := fs.String("data", "./data", "数据目录(SQLite 路径 <data>/agent-system.db)")
|
||||
dsn := fs.String("dsn", "", "PostgreSQL 档 DSN(空 = 嵌入式 SQLite)")
|
||||
instanceID := fs.String("instance-id", "", "服务实例 ID(manifest.instance_id,形如 svc-<group>-…)")
|
||||
dbref := fs.String("dbref", "", "数据库锚点(空 = sqlite 或 pg)")
|
||||
fs.Parse(args[1:])
|
||||
if *instanceID == "" {
|
||||
fmt.Fprintln(os.Stderr, "identity: --instance-id 必填")
|
||||
return 2
|
||||
}
|
||||
ref := *dbref
|
||||
if ref == "" {
|
||||
if *dsn != "" {
|
||||
ref = "pg"
|
||||
} else {
|
||||
ref = "sqlite"
|
||||
}
|
||||
}
|
||||
switch sub {
|
||||
case "ensure":
|
||||
if err := app.EnsureSystemIdentity(*dsn, *data, *instanceID, ref); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "identity ensure:", err)
|
||||
return 1
|
||||
}
|
||||
fmt.Printf("ensured instance_id=%s dbref=%s\n", *instanceID, ref)
|
||||
return 0
|
||||
case "verify":
|
||||
if err := app.VerifySystemIdentity(*dsn, *data, *instanceID, ref); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "identity verify:", err)
|
||||
return 1
|
||||
}
|
||||
fmt.Printf("verified instance_id=%s dbref=%s\n", *instanceID, ref)
|
||||
return 0
|
||||
default:
|
||||
fmt.Fprintln(os.Stderr, "identity: 未知子命令", sub)
|
||||
return 2
|
||||
}
|
||||
}
|
||||
|
||||
@@ -291,6 +291,8 @@ var copyOrder = []string{
|
||||
"secret_keys", "settings_snapshots", "regen_pending", "context_entries",
|
||||
"user_commands", "pending_notices", "recovery_holds",
|
||||
"surfaces", "surface_revisions", "surface_actions", "plugin_activation",
|
||||
// U1 I3:服务实例 DB 身份(无外键;迁移须保留以免目标库丢锚)
|
||||
"agent_system_identity",
|
||||
}
|
||||
|
||||
const controlDDL = `
|
||||
|
||||
@@ -199,6 +199,21 @@ func Run(opt Options) error {
|
||||
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)
|
||||
}
|
||||
}
|
||||
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)",
|
||||
@@ -815,6 +830,20 @@ func RevokeAPIKey(storageDSN, dataDir, owner, name string) (bool, error) {
|
||||
return st.RevokeAPIKeyByName(owner, name)
|
||||
}
|
||||
|
||||
// EnsureSystemIdentity CLI:显式建立库内服务身份(svc init / I2→I3 引导)。
|
||||
func EnsureSystemIdentity(storageDSN, dataDir, instanceID, dbref string) error {
|
||||
return withStore(storageDSN, dataDir, func(st *store.Store) error {
|
||||
return st.EnsureSystemIdentity(instanceID, dbref)
|
||||
})
|
||||
}
|
||||
|
||||
// VerifySystemIdentity CLI:校验库内服务身份(svc start 前;失败不插入)。
|
||||
func VerifySystemIdentity(storageDSN, dataDir, instanceID, dbref string) error {
|
||||
return withStore(storageDSN, dataDir, func(st *store.Store) error {
|
||||
return st.VerifySystemIdentity(instanceID, dbref)
|
||||
})
|
||||
}
|
||||
|
||||
// cachedInt 带短 TTL 的整型缓存读(限流值热读取免打库)。
|
||||
func cachedInt(ttl time.Duration, fn func() int) func() int {
|
||||
var mu sync.Mutex
|
||||
|
||||
@@ -712,6 +712,29 @@ func init() {
|
||||
`ALTER TABLE knowledge_vectors DROP COLUMN content_hash`,
|
||||
`ALTER TABLE knowledge_vectors DROP COLUMN model`,
|
||||
}}},
|
||||
// U1 I3:服务实例 DB 身份 marker(设计 docs/recovery/u1-identity-manifest.md §5.2)。
|
||||
// 全表至多一行由 Ensure/Verify 强制;与 runs.instance_id(进程标识)无关。
|
||||
migration{version: 26, name: "agent_system_identity",
|
||||
up: map[string][]string{"sqlite": {
|
||||
`CREATE TABLE IF NOT EXISTS agent_system_identity (
|
||||
instance_id TEXT NOT NULL PRIMARY KEY,
|
||||
dbref TEXT NOT NULL,
|
||||
schema_version INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
)`,
|
||||
}, "postgres": {
|
||||
`CREATE TABLE IF NOT EXISTS agent_system_identity (
|
||||
instance_id TEXT NOT NULL PRIMARY KEY,
|
||||
dbref TEXT NOT NULL,
|
||||
schema_version INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
)`,
|
||||
}},
|
||||
down: map[string][]string{"sqlite": {
|
||||
`DROP TABLE IF EXISTS agent_system_identity`,
|
||||
}, "postgres": {
|
||||
`DROP TABLE IF EXISTS agent_system_identity`,
|
||||
}}},
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// ErrSystemIdentityMismatch 库内身份与期望不符(错库/错实例/多行)——启动必须失败。
|
||||
var ErrSystemIdentityMismatch = errors.New("system identity mismatch")
|
||||
|
||||
// ErrSystemIdentityMissing 表中无身份行——须先 Ensure(svc init / 首次建立),禁止静默插入。
|
||||
var ErrSystemIdentityMissing = errors.New("system identity missing")
|
||||
|
||||
// SystemIdentity 是 agent_system_identity 表的一行(一库一服务身份)。
|
||||
type SystemIdentity struct {
|
||||
InstanceID string
|
||||
DBRef string
|
||||
SchemaVersion int
|
||||
CreatedAt string
|
||||
}
|
||||
|
||||
// EnsureSystemIdentity 仅用于显式建立路径:表空则插入;已存在且一致则幂等;
|
||||
// 不一致或多余行 fail-closed。绝不在常规 start 路径调用以「补写」掩盖漂移。
|
||||
func (s *Store) EnsureSystemIdentity(instanceID, dbref string) error {
|
||||
if instanceID == "" {
|
||||
return fmt.Errorf("EnsureSystemIdentity: instance_id 为空")
|
||||
}
|
||||
if dbref == "" {
|
||||
dbref = "sqlite"
|
||||
}
|
||||
row, n, err := s.readSystemIdentity()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n > 1 {
|
||||
return fmt.Errorf("%w: agent_system_identity 有 %d 行(期望至多 1)", ErrSystemIdentityMismatch, n)
|
||||
}
|
||||
if n == 1 {
|
||||
if row.InstanceID != instanceID || row.DBRef != dbref {
|
||||
return fmt.Errorf("%w: 库内 instance_id=%q dbref=%q ≠ 期望 %q / %q",
|
||||
ErrSystemIdentityMismatch, row.InstanceID, row.DBRef, instanceID, dbref)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
ver := latestMigrationVersion()
|
||||
_, err = s.db.Exec(s.q(`INSERT INTO agent_system_identity(instance_id, dbref, schema_version, created_at) VALUES(?,?,?,?)`),
|
||||
instanceID, dbref, ver, nowISO())
|
||||
if err != nil {
|
||||
return fmt.Errorf("EnsureSystemIdentity insert: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// VerifySystemIdentity 常规启动校验:必须恰有一行且与期望一致;缺失不插入。
|
||||
func (s *Store) VerifySystemIdentity(instanceID, dbref string) error {
|
||||
if instanceID == "" {
|
||||
return fmt.Errorf("VerifySystemIdentity: instance_id 为空")
|
||||
}
|
||||
if dbref == "" {
|
||||
dbref = "sqlite"
|
||||
}
|
||||
row, n, err := s.readSystemIdentity()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n == 0 {
|
||||
return fmt.Errorf("%w: agent_system_identity 无行(先 svc init / identity ensure,禁止静默建空身份)", ErrSystemIdentityMissing)
|
||||
}
|
||||
if n > 1 {
|
||||
return fmt.Errorf("%w: agent_system_identity 有 %d 行(期望至多 1)", ErrSystemIdentityMismatch, n)
|
||||
}
|
||||
if row.InstanceID != instanceID || row.DBRef != dbref {
|
||||
return fmt.Errorf("%w: 库内 instance_id=%q dbref=%q ≠ 期望 %q / %q",
|
||||
ErrSystemIdentityMismatch, row.InstanceID, row.DBRef, instanceID, dbref)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) readSystemIdentity() (SystemIdentity, int, error) {
|
||||
rows, err := s.db.Query(`SELECT instance_id, dbref, schema_version, created_at FROM agent_system_identity`)
|
||||
if err != nil {
|
||||
return SystemIdentity{}, 0, fmt.Errorf("read agent_system_identity: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out SystemIdentity
|
||||
n := 0
|
||||
for rows.Next() {
|
||||
n++
|
||||
var id, ref, created string
|
||||
var ver int
|
||||
if err := rows.Scan(&id, &ref, &ver, &created); err != nil {
|
||||
return SystemIdentity{}, 0, err
|
||||
}
|
||||
if n == 1 {
|
||||
out = SystemIdentity{InstanceID: id, DBRef: ref, SchemaVersion: ver, CreatedAt: created}
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return SystemIdentity{}, 0, err
|
||||
}
|
||||
return out, n, nil
|
||||
}
|
||||
|
||||
// GetSystemIdentity 读取唯一身份行(无行返回 sql.ErrNoRows)。
|
||||
func (s *Store) GetSystemIdentity() (SystemIdentity, error) {
|
||||
row, n, err := s.readSystemIdentity()
|
||||
if err != nil {
|
||||
return SystemIdentity{}, err
|
||||
}
|
||||
if n == 0 {
|
||||
return SystemIdentity{}, sql.ErrNoRows
|
||||
}
|
||||
if n > 1 {
|
||||
return SystemIdentity{}, fmt.Errorf("%w: %d rows", ErrSystemIdentityMismatch, n)
|
||||
}
|
||||
return row, nil
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSystemIdentityEnsureVerifySQLite(t *testing.T) {
|
||||
st, err := OpenSQLite(filepath.Join(t.TempDir(), "id.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
if err := st.VerifySystemIdentity("svc-a-aaaaaa", "sqlite"); !errors.Is(err, ErrSystemIdentityMissing) {
|
||||
t.Fatalf("empty verify: want missing, got %v", err)
|
||||
}
|
||||
if err := st.EnsureSystemIdentity("svc-a-aaaaaa", "sqlite"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.EnsureSystemIdentity("svc-a-aaaaaa", "sqlite"); err != nil {
|
||||
t.Fatalf("idempotent ensure: %v", err)
|
||||
}
|
||||
if err := st.VerifySystemIdentity("svc-a-aaaaaa", "sqlite"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := st.EnsureSystemIdentity("svc-b-bbbbbb", "sqlite"); !errors.Is(err, ErrSystemIdentityMismatch) {
|
||||
t.Fatalf("wrong id ensure: want mismatch, got %v", err)
|
||||
}
|
||||
if err := st.VerifySystemIdentity("svc-a-aaaaaa", "pg:other"); !errors.Is(err, ErrSystemIdentityMismatch) {
|
||||
t.Fatalf("wrong dbref verify: want mismatch, got %v", err)
|
||||
}
|
||||
row, err := st.GetSystemIdentity()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if row.InstanceID != "svc-a-aaaaaa" || row.DBRef != "sqlite" {
|
||||
t.Fatalf("row=%+v", row)
|
||||
}
|
||||
if row.SchemaVersion != latestMigrationVersion() {
|
||||
t.Fatalf("schema_version=%d want %d", row.SchemaVersion, latestMigrationVersion())
|
||||
}
|
||||
}
|
||||
|
||||
func TestSystemIdentityMigrationCreatesTable(t *testing.T) {
|
||||
st, err := OpenSQLite(filepath.Join(t.TempDir(), "mig.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
versions, err := st.MigrationStatus()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
found := false
|
||||
for _, v := range versions {
|
||||
if v == 26 {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("migration v26 not applied: %v", versions)
|
||||
}
|
||||
var name string
|
||||
if err := st.DB().QueryRow(`SELECT name FROM sqlite_master WHERE type='table' AND name='agent_system_identity'`).Scan(&name); err != nil || name != "agent_system_identity" {
|
||||
t.Fatalf("table missing: name=%q err=%v", name, err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user