This is an automated email from the ASF dual-hosted git repository. hsluoyz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit ad0af43be220f5e5e7044f2f01694e1fb5c8c72e Author: Yang Luo <[email protected]> AuthorDate: Fri Sep 4 11:46:38 2026 +0800 feat(monitor): keep agent activity records in the database --- agentmonitor/monitor.go | 12 +- agentmonitor/record.go | 10 +- agentmonitor/sink.go | 236 ++----------------------- conf/app.conf | 4 +- conf/conf.go | 6 +- controllers/agent.go | 21 +-- controllers/setting.go | 1 - main.go | 7 +- object/agent_record.go | 346 +++++++++++++++++++++++++++++++++++++ object/ormer.go | 7 + web/src/locales/en/data.json | 4 +- web/src/locales/zh/data.json | 4 +- web/src/pages/AgentRecordsPage.tsx | 4 +- web/src/types.ts | 2 +- 14 files changed, 404 insertions(+), 260 deletions(-) diff --git a/agentmonitor/monitor.go b/agentmonitor/monitor.go index aa1d6d2..a9f60af 100644 --- a/agentmonitor/monitor.go +++ b/agentmonitor/monitor.go @@ -41,10 +41,10 @@ var monitorConfig = struct { pollInterval time.Duration }{stateDir: DefaultStateDir, pollInterval: DefaultPollInterval} -// Configure selects the directory used for declaration and cursor JSON files, -// how often local logs are rescanned, and how many records the live window -// holds. It must be called before Start when the defaults are unsuitable. -func Configure(stateDir string, pollInterval time.Duration, recordCapacity int) { +// Configure selects the directory used for declaration and cursor JSON files +// and how often local logs are rescanned. It must be called before Start when +// the defaults are unsuitable. +func Configure(stateDir string, pollInterval time.Duration) { monitorConfig.Lock() if strings.TrimSpace(stateDir) == "" { monitorConfig.stateDir = DefaultStateDir @@ -57,8 +57,6 @@ func Configure(stateDir string, pollInterval time.Duration, recordCapacity int) monitorConfig.pollInterval = pollInterval } monitorConfig.Unlock() - - SetRecordCapacity(recordCapacity) } func monitorStatePath(name string) string { @@ -81,7 +79,7 @@ func Start() error { return errors.Join(startCoworkMonitor(), codexMonitor.start(), openAgentMonitor.start()) } -// Stop stops every local tailer. It does not persist behaviour records. +// Stop stops every local tailer. func Stop() { stopCoworkMonitor() codexMonitor.stopMonitor() diff --git a/agentmonitor/record.go b/agentmonitor/record.go index 68390f0..5a20d21 100644 --- a/agentmonitor/record.go +++ b/agentmonitor/record.go @@ -18,7 +18,6 @@ import ( "time" "github.com/apache/casbin-gateway/auditutil" - "github.com/google/uuid" ) // IngestTokenHeader carries the per-installation secret that a hook or MCP @@ -26,10 +25,10 @@ import ( // packages and the patcher packages can share it without importing each other. const IngestTokenHeader = "X-Casbin-Gateway-Agent-Token" -// Record is one observed agent behaviour. It is deliberately independent from -// Gateway's HTTP traffic record and is retained only in the live monitor store. +// Record is one observed agent behaviour, as the tailers and the hooks report +// it. It is deliberately independent from Gateway's HTTP traffic record, and +// carries no identity of its own: the sink is what gives a stored record one. type Record struct { - Id string `json:"id"` CreatedTime string `json:"createdTime"` Agent string `json:"agent"` @@ -56,9 +55,6 @@ type Record struct { } func normalizeRecord(record *Record) { - if record.Id == "" { - record.Id = uuid.NewString() - } if record.CreatedTime == "" { record.CreatedTime = time.Now().Format(time.RFC3339Nano) } diff --git a/agentmonitor/sink.go b/agentmonitor/sink.go index 1659a3b..48ef4b4 100644 --- a/agentmonitor/sink.go +++ b/agentmonitor/sink.go @@ -14,232 +14,34 @@ package agentmonitor -import ( - "sort" - "strings" - "sync" - "time" -) +import "sync" -// DefaultRecordCapacity is the live diagnostic window retained by Gateway when -// no capacity is configured. Records live only in memory, so this value bounds -// how much a busy host can retain: each record may hold up to -// auditutil.MaxPayloadBytes of sanitized payload. -const DefaultRecordCapacity = 1000 +// RecordSink is where observed behaviour is kept. Storage lives outside this +// package so that tailing an agent's logs does not depend on the database. +type RecordSink func(*Record) -// RecordQuery narrows the live diagnostic window. Empty fields match all -// values and a non-positive Limit returns every matching record. -type RecordQuery struct { - Agent string - EventType string - Outcome string - Session string - Limit int +var recordSink struct { + sync.RWMutex + sink RecordSink } -// SessionSummary is the session-level view derived from the live records. -type SessionSummary struct { - SessionKey string `json:"sessionKey"` - Agent string `json:"agent"` - Title string `json:"title"` - RecordCount int `json:"recordCount"` - FirstTime string `json:"firstTime"` - LastTime string `json:"lastTime"` +// SetRecordSink installs the store every record is written to. A record +// observed before one is installed is dropped. +func SetRecordSink(sink RecordSink) { + recordSink.Lock() + recordSink.sink = sink + recordSink.Unlock() } -type storedRecord struct { - record Record - sequence uint64 -} - -// Store keeps a bounded, thread-safe window of recent monitoring records. -type Store struct { - mutex sync.RWMutex - entries []storedRecord - next int - count int - sequence uint64 -} - -var recordStore = newStore(DefaultRecordCapacity) - -func newStore(capacity int) *Store { - if capacity <= 0 { - capacity = DefaultRecordCapacity - } - return &Store{entries: make([]storedRecord, capacity)} -} - -// SetRecordCapacity resizes the live window, keeping the most recent records -// that still fit. It is a no-op when the capacity is unchanged. -func SetRecordCapacity(capacity int) { - recordStore.SetCapacity(capacity) -} - -// SetCapacity resizes the ring buffer in place, preserving arrival order. -func (store *Store) SetCapacity(capacity int) { - if capacity <= 0 { - capacity = DefaultRecordCapacity - } - - store.mutex.Lock() - defer store.mutex.Unlock() - if capacity == len(store.entries) { - return - } - - retained := store.snapshotLocked() - if len(retained) > capacity { - retained = retained[len(retained)-capacity:] - } - store.entries = make([]storedRecord, capacity) - copy(store.entries, retained) - store.count = len(retained) - store.next = store.count % capacity -} - -// AddRecord appends one event to Gateway's process-local monitoring window. +// AddRecord normalizes one observed behaviour and hands it to the sink. func AddRecord(record *Record) { - recordStore.Add(record) -} - -// Add normalizes and appends one record, replacing the oldest record once the -// capacity is reached. -func (store *Store) Add(record *Record) { value := *record normalizeRecord(&value) - store.mutex.Lock() - store.sequence++ - store.entries[store.next] = storedRecord{record: value, sequence: store.sequence} - store.next = (store.next + 1) % len(store.entries) - if store.count < len(store.entries) { - store.count++ - } - store.mutex.Unlock() -} - -// ListRecords returns matching records ordered by their reported time, newest -// first. Ties preserve their arrival order. -func ListRecords(query RecordQuery) []Record { - return recordStore.List(query) -} - -// List returns matching records ordered by reported time, newest first. -func (store *Store) List(query RecordQuery) []Record { - store.mutex.RLock() - matches := make([]storedRecord, 0, store.count) - for _, entry := range store.snapshotLocked() { - if matchesRecord(entry.record, query) { - matches = append(matches, entry) - } - } - store.mutex.RUnlock() - - sort.SliceStable(matches, func(left, right int) bool { - leftTime, rightTime := recordTime(matches[left].record), recordTime(matches[right].record) - if leftTime.Equal(rightTime) { - return matches[left].sequence > matches[right].sequence - } - return leftTime.After(rightTime) - }) - if query.Limit > 0 && len(matches) > query.Limit { - matches = matches[:query.Limit] - } - result := make([]Record, len(matches)) - for index, entry := range matches { - result[index] = entry.record - } - return result -} - -// ListSessions returns session summaries newest first. -func ListSessions(query RecordQuery) []SessionSummary { - return recordStore.Sessions(query) -} - -// Sessions groups matching records by their session key. -func (store *Store) Sessions(query RecordQuery) []SessionSummary { - query.Limit = 0 - records := store.List(query) - groups := map[string][]Record{} - order := make([]string, 0) - for _, record := range records { - if record.SessionKey == "" { - continue - } - key := record.Agent + "\x00" + record.SessionKey - if _, found := groups[key]; !found { - order = append(order, key) - } - groups[key] = append(groups[key], record) - } - - result := make([]SessionSummary, 0, len(order)) - for _, key := range order { - group := groups[key] - newest, oldest := group[0], group[len(group)-1] - result = append(result, SessionSummary{ - SessionKey: newest.SessionKey, - Agent: newest.Agent, - Title: sessionTitle(group), - RecordCount: len(group), - FirstTime: oldest.CreatedTime, - LastTime: newest.CreatedTime, - }) - } - return result -} - -func (store *Store) snapshotLocked() []storedRecord { - result := make([]storedRecord, 0, store.count) - start := 0 - if store.count == len(store.entries) { - start = store.next - } - for index := 0; index < store.count; index++ { - result = append(result, store.entries[(start+index)%len(store.entries)]) - } - return result -} - -func matchesRecord(record Record, query RecordQuery) bool { - return matchesFilter(record.Agent, query.Agent) && - matchesFilter(record.EventType, query.EventType) && - matchesFilter(record.Outcome, query.Outcome) && - matchesFilter(record.SessionKey, query.Session) -} - -func matchesFilter(value, filter string) bool { - return filter == "" || strings.EqualFold(value, filter) -} - -func recordTime(record Record) time.Time { - parsed, err := time.Parse(time.RFC3339Nano, record.CreatedTime) - if err != nil { - return time.Time{} - } - return parsed -} - -func sessionTitle(newestFirst []Record) string { - for _, record := range newestFirst { - if record.Title != "" { - return record.Title - } - } - for index := len(newestFirst) - 1; index >= 0; index-- { - record := newestFirst[index] - if record.McpServer != "" { - if record.McpTool != "" { - return record.McpServer + " / " + record.McpTool - } - return record.McpServer - } - if record.ToolName != "" { - return record.ToolName - } + recordSink.RLock() + sink := recordSink.sink + recordSink.RUnlock() + if sink != nil { + sink(&value) } - oldest := newestFirst[len(newestFirst)-1] - return strings.TrimSpace(oldest.EventType + " " + oldest.Action) } diff --git a/conf/app.conf b/conf/app.conf index 5d5266e..0f93c1e 100644 --- a/conf/app.conf +++ b/conf/app.conf @@ -63,8 +63,8 @@ isDemoMode = false ; Agent patch manifests, config backups, ingest tokens and file-tail cursors. agentPatchStateDir = "./data/agent-patches" -; How many agent monitoring records to keep. They are never persisted, so this -; is also the memory bound for the live window. +; How many agent monitoring records to keep. They are stored in the database, +; and the oldest are pruned once the table is over this many rows. agentRecordCapacity = 1000 ; How often to rescan the append-only logs of file-tailed agents, in seconds. agentMonitorPollSeconds = 5 diff --git a/conf/conf.go b/conf/conf.go index dce2299..f6cc65b 100644 --- a/conf/conf.go +++ b/conf/conf.go @@ -313,9 +313,9 @@ func GetAgentPatchStateDir() string { return dir } -// GetAgentRecordCapacity is how many agent monitoring records Gateway keeps in -// memory. Records are never written to disk, so this value alone bounds the -// memory the live window can occupy. +// GetAgentRecordCapacity is how many agent monitoring records Gateway keeps. +// They are stored in the database, and the oldest are pruned once the table is +// over this many rows. func GetAgentRecordCapacity() int { res, err := strconv.Atoi(GetConfigString("agentRecordCapacity")) if err != nil || res <= 0 { diff --git a/controllers/agent.go b/controllers/agent.go index 0b0dc79..c4d124c 100644 --- a/controllers/agent.go +++ b/controllers/agent.go @@ -240,7 +240,7 @@ func (c *ApiController) UnpatchAgent() { c.ResponseOk(agentpatch.StatusOf(target)) } -// GetAgentRecords returns the current process's in-memory agent activity. +// GetAgentRecords returns the stored agent activity. func (c *ApiController) GetAgentRecords() { if c.RequireAdmin() { return @@ -255,7 +255,7 @@ func (c *ApiController) GetAgentRecords() { } limit = parsed } - c.ResponseOk(agentmonitor.ListRecords(agentmonitor.RecordQuery{ + c.ResponseOk(object.GetAgentRecords(object.AgentRecordFilter{ Agent: c.Input().Get("agent"), EventType: c.Input().Get("eventType"), Outcome: c.Input().Get("outcome"), @@ -264,30 +264,21 @@ func (c *ApiController) GetAgentRecords() { })) } -// GetAgentSessions groups the current in-memory records by agent session. The -// optional agent filter is what an agent's own detail page asks for. +// GetAgentSessions groups the stored records by agent session. The optional +// agent filter is what an agent's own detail page asks for. func (c *ApiController) GetAgentSessions() { if c.RequireAdmin() { return } agentId := c.Input().Get("agent") - live := agentmonitor.ListSessions(agentmonitor.RecordQuery{Agent: agentId}) + sessions := object.GetAgentRecordSessions(agentId) // The transcripts on disk are the sessions that already happened, so they // are listed next to the monitored ones rather than only after Patch. - sessions := make([]agenthistory.Session, 0, len(live)) seen := map[string]bool{} - for _, session := range live { + for _, session := range sessions { seen[sessionSeenKey(session.Agent, session.SessionKey)] = true - sessions = append(sessions, agenthistory.Session{ - Agent: session.Agent, - SessionKey: session.SessionKey, - Title: session.Title, - RecordCount: session.RecordCount, - FirstTime: session.FirstTime, - LastTime: session.LastTime, - }) } for _, session := range object.HistoricalSessions(agentId) { if seen[sessionSeenKey(session.Agent, session.SessionKey)] { diff --git a/controllers/setting.go b/controllers/setting.go index fd97713..0a66b1b 100644 --- a/controllers/setting.go +++ b/controllers/setting.go @@ -110,7 +110,6 @@ func applySetting(previous *object.Setting, setting *object.Setting) error { agentmonitor.Configure( conf.GetAgentPatchStateDir(), time.Duration(conf.GetAgentMonitorPollSeconds())*time.Second, - conf.GetAgentRecordCapacity(), ) if setting.LlmRecordMode == conf.LlmRecordOff { diff --git a/main.go b/main.go index e7802c0..3639947 100644 --- a/main.go +++ b/main.go @@ -105,8 +105,12 @@ func main() { agentmonitor.Configure( conf.GetAgentPatchStateDir(), time.Duration(conf.GetAgentMonitorPollSeconds())*time.Second, - conf.GetAgentRecordCapacity(), ) + // Monitoring observes; the database is what keeps what it saw, so records + // outlive the process that collected them. + object.StartAgentRecordWriter() + defer object.StopAgentRecordWriter() + agentmonitor.SetRecordSink(object.AddAgentRecord) if err := agentmonitor.Start(); err != nil { beego.Error("agent monitor could not start:", err) } @@ -125,6 +129,7 @@ func main() { // calls above would have flushed has to be flushed there instead. version.BeforeRestart = func() { agentmonitor.Stop() + object.StopAgentRecordWriter() object.StopLlmRecordWriter() } diff --git a/object/agent_record.go b/object/agent_record.go new file mode 100644 index 0000000..f930f0a --- /dev/null +++ b/object/agent_record.go @@ -0,0 +1,346 @@ +// Copyright 2026 The casbin Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package object + +import ( + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/apache/casbin-gateway/agenthistory" + "github.com/apache/casbin-gateway/agentmonitor" + "github.com/apache/casbin-gateway/conf" + "github.com/beego/beego" + "github.com/xorm-io/xorm" +) + +// agentRecordQueueCapacity bounds the records waiting to be written. A tailer +// must never wait on the database, so a full queue drops instead. +const agentRecordQueueCapacity = 1000 + +// AgentRecord is one observed agent behaviour, reported by a log tailer, a hook +// or the MCP recorder. Unlike LlmRecord it says nothing about what Gateway +// relayed: an agent talking to its vendor directly is recorded here all the same. +type AgentRecord struct { + Id int64 `xorm:"int notnull pk autoincr" json:"id"` + CreatedTime string `xorm:"varchar(100) notnull" json:"createdTime"` + // SortKey is the created time in Unix nanoseconds. A tailer reports times in + // whichever zone the log it reads is written in, so the text column alone + // does not order rows the way a clock would. + SortKey int64 `xorm:"bigint index" json:"-"` + + Agent string `xorm:"varchar(201) index" json:"agent"` + AgentPath string `xorm:"varchar(500)" json:"agentPath,omitempty"` + ClientIp string `xorm:"varchar(100)" json:"clientIp,omitempty"` + User string `xorm:"varchar(201)" json:"user,omitempty"` + + EventType string `xorm:"varchar(50) index" json:"eventType"` + Action string `xorm:"varchar(100)" json:"action"` + Outcome string `xorm:"varchar(50) index" json:"outcome,omitempty"` + SessionKey string `xorm:"varchar(201) index" json:"sessionKey,omitempty"` + Title string `xorm:"varchar(500)" json:"title,omitempty"` + + PromptId string `xorm:"varchar(201)" json:"promptId,omitempty"` + ToolUseId string `xorm:"varchar(201)" json:"toolUseId,omitempty"` + ToolName string `xorm:"varchar(201)" json:"toolName,omitempty"` + McpServer string `xorm:"varchar(201)" json:"mcpServer,omitempty"` + McpTool string `xorm:"varchar(201)" json:"mcpTool,omitempty"` + Model string `xorm:"varchar(255)" json:"model,omitempty"` + DurationMs int64 `xorm:"bigint" json:"durationMs,omitempty"` + + Object string `xorm:"mediumtext" json:"object,omitempty"` + Detail string `xorm:"mediumtext" json:"detail,omitempty"` +} + +// AgentRecordFilter narrows the record list. Empty fields match everything and +// a non-positive Limit returns every matching record. +type AgentRecordFilter struct { + Agent string + EventType string + Outcome string + Session string + Limit int +} + +// agentSessionColumns are what grouping records into sessions reads. A record +// may carry tens of kilobytes of payload, so the summary never selects one. +var agentSessionColumns = []string{ + "agent", "session_key", "title", "created_time", + "event_type", "action", "tool_name", "mcp_server", "mcp_tool", +} + +type agentRecordWriter struct { + mutex sync.RWMutex + queue chan *AgentRecord + done chan struct{} + + dropped atomic.Int64 + dropLogTime atomic.Int64 +} + +var agentWriter agentRecordWriter + +// StartAgentRecordWriter starts the goroutine that owns every agent record +// write, so a tailer or a hook never blocks on the database. +func StartAgentRecordWriter() { + agentWriter.mutex.Lock() + defer agentWriter.mutex.Unlock() + if agentWriter.queue != nil { + return + } + agentWriter.queue = make(chan *AgentRecord, agentRecordQueueCapacity) + agentWriter.done = make(chan struct{}) + go agentWriter.run(agentWriter.queue, agentWriter.done) +} + +// StopAgentRecordWriter drains the records already accepted. +func StopAgentRecordWriter() { + agentWriter.mutex.Lock() + queue, done := agentWriter.queue, agentWriter.done + agentWriter.queue, agentWriter.done = nil, nil + agentWriter.mutex.Unlock() + + if queue == nil { + return + } + close(queue) + <-done +} + +// AddAgentRecord queues one observed behaviour. It is the sink agentmonitor +// reports to, and is best effort: a full queue drops the record rather than +// holding a tailer up. +func AddAgentRecord(record *agentmonitor.Record) { + if record == nil { + return + } + + agentWriter.mutex.RLock() + defer agentWriter.mutex.RUnlock() + if agentWriter.queue == nil { + return + } + select { + case agentWriter.queue <- newAgentRecord(record): + default: + agentWriter.noteDrop() + } +} + +func newAgentRecord(record *agentmonitor.Record) *AgentRecord { + return &AgentRecord{ + CreatedTime: record.CreatedTime, + SortKey: agentRecordSortKey(record.CreatedTime), + Agent: record.Agent, + AgentPath: record.AgentPath, + ClientIp: record.ClientIp, + User: record.User, + EventType: record.EventType, + Action: record.Action, + Outcome: record.Outcome, + SessionKey: record.SessionKey, + Title: record.Title, + PromptId: record.PromptId, + ToolUseId: record.ToolUseId, + ToolName: record.ToolName, + McpServer: record.McpServer, + McpTool: record.McpTool, + Model: record.Model, + DurationMs: record.DurationMs, + Object: record.Object, + Detail: record.Detail, + } +} + +func agentRecordSortKey(createdTime string) int64 { + parsed, err := time.Parse(time.RFC3339Nano, createdTime) + if err != nil { + return time.Now().UnixNano() + } + return parsed.UnixNano() +} + +func (writer *agentRecordWriter) noteDrop() { + dropped := writer.dropped.Add(1) + last := writer.dropLogTime.Load() + now := time.Now().Unix() + if now-last < 60 || !writer.dropLogTime.CompareAndSwap(last, now) { + return + } + beego.Error("agent record queue is full, dropped records so far:", dropped) +} + +func (writer *agentRecordWriter) run(queue chan *AgentRecord, done chan struct{}) { + defer close(done) + + writer.prune() + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case record, ok := <-queue: + if !ok { + return + } + writer.write(record) + case <-ticker.C: + writer.prune() + } + } +} + +func (writer *agentRecordWriter) write(record *AgentRecord) { + if ormer == nil || ormer.Engine == nil { + return + } + if _, err := ormer.Engine.Insert(record); err != nil { + beego.Error("agent record write failed:", err) + } +} + +// prune keeps the newest agentRecordCapacity rows. The records are a monitoring +// window rather than an audit trail, so the cap is the only retention rule. +func (writer *agentRecordWriter) prune() { + if ormer == nil || ormer.Engine == nil { + return + } + + count, err := ormer.Engine.Count(&AgentRecord{}) + if err != nil { + beego.Error("agent record retention count failed:", err) + return + } + maximum := int64(conf.GetAgentRecordCapacity()) + if count <= maximum { + return + } + + oldest := []AgentRecord{} + if err := ormer.Engine.Cols("id").Asc("id").Limit(int(count - maximum)).Find(&oldest); err != nil { + beego.Error("agent record retention lookup failed:", err) + return + } + if len(oldest) == 0 { + return + } + ids := make([]int64, 0, len(oldest)) + for _, record := range oldest { + ids = append(ids, record.Id) + } + if _, err := ormer.Engine.In("id", ids).Delete(&AgentRecord{}); err != nil { + beego.Error("agent record retention deletion failed:", err) + } +} + +// GetAgentRecords returns matching records newest first. +func GetAgentRecords(filter AgentRecordFilter) []*AgentRecord { + records := []*AgentRecord{} + if ormer == nil || ormer.Engine == nil { + return records + } + + session := agentRecordSession(filter).Desc("sort_key").Desc("id") + if filter.Limit > 0 { + session = session.Limit(filter.Limit) + } + if err := session.Find(&records); err != nil { + beego.Error("agent record lookup failed:", err) + return []*AgentRecord{} + } + return records +} + +// GetAgentRecordSessions groups the stored records into the sessions they +// belong to, newest first. It is the monitored half of the Sessions page, the +// other half being the transcripts on disk. +func GetAgentRecordSessions(agentId string) []agenthistory.Session { + sessions := []agenthistory.Session{} + if ormer == nil || ormer.Engine == nil { + return sessions + } + + records := []*AgentRecord{} + query := agentRecordSession(AgentRecordFilter{Agent: agentId}).Cols(agentSessionColumns...) + if err := query.And("session_key <> ?", "").Desc("sort_key").Desc("id").Find(&records); err != nil { + beego.Error("agent session lookup failed:", err) + return sessions + } + + groups := map[string][]*AgentRecord{} + order := []string{} + for _, record := range records { + key := record.Agent + "\x00" + record.SessionKey + if _, found := groups[key]; !found { + order = append(order, key) + } + groups[key] = append(groups[key], record) + } + + for _, key := range order { + group := groups[key] + newest, oldest := group[0], group[len(group)-1] + sessions = append(sessions, agenthistory.Session{ + Agent: newest.Agent, + SessionKey: newest.SessionKey, + Title: agentSessionTitle(group), + RecordCount: len(group), + FirstTime: oldest.CreatedTime, + LastTime: newest.CreatedTime, + }) + } + return sessions +} + +func agentRecordSession(filter AgentRecordFilter) *xorm.Session { + session := ormer.Engine.NewSession() + if filter.Agent != "" { + session = session.And("lower(agent) = ?", strings.ToLower(filter.Agent)) + } + if filter.EventType != "" { + session = session.And("lower(event_type) = ?", strings.ToLower(filter.EventType)) + } + if filter.Outcome != "" { + session = session.And("lower(outcome) = ?", strings.ToLower(filter.Outcome)) + } + if filter.Session != "" { + session = session.And("lower(session_key) = ?", strings.ToLower(filter.Session)) + } + return session +} + +// agentSessionTitle names a session by what it was seen doing: the title the +// agent reported, or failing that the first thing it called. +func agentSessionTitle(newestFirst []*AgentRecord) string { + for _, record := range newestFirst { + if record.Title != "" { + return record.Title + } + } + for index := len(newestFirst) - 1; index >= 0; index-- { + record := newestFirst[index] + if record.McpServer != "" { + if record.McpTool != "" { + return record.McpServer + " / " + record.McpTool + } + return record.McpServer + } + if record.ToolName != "" { + return record.ToolName + } + } + oldest := newestFirst[len(newestFirst)-1] + return strings.TrimSpace(oldest.EventType + " " + oldest.Action) +} diff --git a/object/ormer.go b/object/ormer.go index fc51463..a26aba8 100644 --- a/object/ormer.go +++ b/object/ormer.go @@ -288,6 +288,13 @@ func (a *Ormer) createTable() { panic(err) } + // What the monitored agents on this host were seen doing, which is what the + // Activity page is made of. + err = a.Engine.Sync2(new(AgentRecord)) + if err != nil { + panic(err) + } + // What an active probe found the last few times it asked a provider's // upstream the questions the passive records cannot answer. err = a.Engine.Sync2(new(ProviderProbe)) diff --git a/web/src/locales/en/data.json b/web/src/locales/en/data.json index f32cea6..fe1fa13 100644 --- a/web/src/locales/en/data.json +++ b/web/src/locales/en/data.json @@ -197,7 +197,7 @@ "Read many files at once": "Read many files at once", "Reading the project": "Reading the project", "Records": "Records", - "Records are kept in memory only and are lost when Gateway restarts": "Records are kept in memory only and are lost when Gateway restarts", + "Records are stored in the database and survive a Gateway restart": "Records are stored in the database and survive a Gateway restart", "Remember something": "Remember something", "Remove instance": "Remove instance", "Remove instance hint": "Gateway forgets this instance. Its state directory, and the account signed in to it, are left on disk.", @@ -1042,7 +1042,7 @@ "Added": "Added", "Agent poll seconds": "Agent poll interval, in seconds", "Agent record capacity": "Agent records kept", - "Agent record capacity hint": "Agent records are never written to disk, so this also bounds the memory they use.", + "Agent record capacity hint": "The oldest records are pruned once the table is over this many rows.", "Agent routing": "Agent routing", "Agent state dir": "Agent state directory", "Agent state dir hint": "Patch manifests, config backups, ingest tokens and file-tail cursors.", diff --git a/web/src/locales/zh/data.json b/web/src/locales/zh/data.json index b695a8c..d443060 100644 --- a/web/src/locales/zh/data.json +++ b/web/src/locales/zh/data.json @@ -197,7 +197,7 @@ "Read many files at once": "一次读取多个文件", "Reading the project": "读取项目", "Records": "记录数", - "Records are kept in memory only and are lost when Gateway restarts": "记录仅保存在内存中，Gateway 重启后会丢失", + "Records are stored in the database and survive a Gateway restart": "记录保存在数据库中，Gateway 重启后依然存在", "Remember something": "记住信息", "Remove instance": "移除实例", "Remove instance hint": "Gateway 将不再管理该实例。它的状态目录以及其中登录的账号仍保留在磁盘上。", @@ -1114,7 +1114,7 @@ "Added": "新增", "Agent poll seconds": "Agent 轮询间隔（秒）", "Agent record capacity": "Agent 记录条数", - "Agent record capacity hint": "Agent 记录不会落盘，因此这个值也决定了它们占用的内存上限。", + "Agent record capacity hint": "记录条数超过这个值后，最旧的记录会被清理。", "Agent routing": "代理路由", "Agent state dir": "Agent 状态目录", "Agent state dir hint": "补丁清单、配置备份、上报令牌和文件读取位置。", diff --git a/web/src/pages/AgentRecordsPage.tsx b/web/src/pages/AgentRecordsPage.tsx index 96d6b6a..4daf118 100644 --- a/web/src/pages/AgentRecordsPage.tsx +++ b/web/src/pages/AgentRecordsPage.tsx @@ -33,7 +33,7 @@ import {Label} from "@/components/ui/label"; import {Switch} from "@/components/ui/switch"; import type {Account, AgentRecord} from "@/types"; -// The server keeps a bounded in-memory window; these are the slices of it the +// The server keeps a bounded window of rows; these are the slices of it the // page can ask for. Without a control here the UI could only ever reach the // first 200 of the records the backend was holding. const limitOptions = [200, 500, 1000, 5000]; @@ -268,7 +268,7 @@ export default function AgentRecordsPage({account}: {account: Account}) { <PageContainer> <PageHeader title={i18next.t("agent:Agent Records")} - description={i18next.t("agent:Records are kept in memory only and are lost when Gateway restarts")} + description={i18next.t("agent:Records are stored in the database and survive a Gateway restart")} actions={ <> <Label className="text-sm font-normal"> diff --git a/web/src/types.ts b/web/src/types.ts index c7ef833..9291c12 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -340,7 +340,7 @@ export interface ProviderHealth { } export interface AgentRecord { - id: string; + id: number; createdTime: string; agent: string; agentPath?: string; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
