From a7c34a608770329fab74d854592a2f310a4eeba0 Mon Sep 17 00:00:00 2001 From: viettranx Date: Mon, 23 Mar 2026 23:41:27 +0700 Subject: [PATCH] =?UTF-8?q?fix:=20address=204=20bugs=20from=20issue=20#294?= =?UTF-8?q?=20=E2=80=94=20context=20propagation,=20cron=20reliability,=20c?= =?UTF-8?q?rash=20safety?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Stale user_context_files after agent file update (#294.1) - Added PropagateContextFile() to AgentStore — single CTE query copies agent-level file to all existing user instances - Added `propagate` param to agents.files.set WS handler - Response includes `propagated` count 2. Cron session reuse — tool errors block future runs (#294.2) - Reset + Save cron session before each run - Each run starts clean, persisted to DB to survive restarts 3. Write-behind cache risk — crash loses messages (#294.3) - Added periodic checkpoint flush every 5 tool iterations - Messages persisted to session incrementally during long runs - Token calibration tracks checkpointed messages correctly 4. Silent cron delivery failures (#294.4) - Added slog.Warn when deliver=true but channel/chatID missing Closes #294 --- cmd/gateway.go | 2 +- cmd/gateway_cron.go | 12 ++++++++- internal/agent/loop.go | 18 ++++++++++++- internal/bootstrap/seed_store_test.go | 3 +++ .../methods/agents_create_owner_test.go | 3 +++ internal/gateway/methods/agents_files.go | 20 +++++++++++--- internal/store/agent_store.go | 4 +++ internal/store/pg/agents_context.go | 26 +++++++++++++++++++ .../tools/context_file_interceptor_test.go | 3 +++ 9 files changed, 85 insertions(+), 6 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index 9d523ea9..57c9a5f6 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -789,7 +789,7 @@ func runGateway() { defer sched.Stop() // Start cron service with job handler (routes through scheduler's cron lane) - pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr)) + pgStores.Cron.SetOnJob(makeCronJobHandler(sched, msgBus, cfg, channelMgr, pgStores.Sessions)) pgStores.Cron.SetOnEvent(func(event store.CronEvent) { server.BroadcastEvent(*protocol.NewEvent(protocol.EventCron, event)) }) diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index 1abb565b..12e33fd3 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -3,6 +3,7 @@ package cmd import ( "context" "fmt" + "log/slog" "strings" "github.com/nextlevelbuilder/goclaw/internal/agent" @@ -21,7 +22,7 @@ import ( // Safe because cron jobs only fire after Start(), well after this is set. var cronHeartbeatWakeFn func(agentID string) -func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager) func(job *store.CronJob) (*store.CronJobResult, error) { +func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager, sessionMgr store.SessionStore) func(job *store.CronJob) (*store.CronJobResult, error) { return func(job *store.CronJob) (*store.CronJobResult, error) { agentID := job.AgentID if agentID == "" { @@ -63,6 +64,12 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg // Build context with tenant scope so agent loop events are scoped correctly. cronCtx := store.WithTenantID(context.Background(), job.TenantID) + // Reset session before each cron run to prevent tool errors from previous + // runs from polluting the context and blocking future executions (#294). + // Save() persists the empty session to DB so stale data won't reload after restart. + sessionMgr.Reset(cronCtx, sessionKey) + sessionMgr.Save(cronCtx, sessionKey) + // Schedule through cron lane — scheduler handles agent resolution and concurrency outCh := sched.Schedule(cronCtx, scheduler.LaneCron, agent.RunRequest{ SessionKey: sessionKey, @@ -99,6 +106,9 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg } appendMediaToOutbound(&outMsg, result.Media) msgBus.PublishOutbound(outMsg) + } else if job.Payload.Deliver { + slog.Warn("cron: delivery configured but channel/chatID missing — output discarded", + "job_id", job.ID, "job_name", job.Name, "channel", job.Payload.Channel, "to", job.Payload.To) } cronResult := &store.CronJobResult{ diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 7b5963bc..435a23f0 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -512,6 +512,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) var deliverables []string // actual content from tool outputs (for team task results) var blockReplies int // count of block.reply events emitted (for dedup in consumer) var lastBlockReply string // last block reply content + var checkpointFlushedMsgs int // messages flushed mid-run for crash safety (#294) // Mid-loop compaction: summarize in-memory messages when context exceeds threshold. // Uses same config as maybeSummarize (contextWindow * historyShare). @@ -1330,6 +1331,21 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) pendingMsgs = append(pendingMsgs, forSession...) } + // Periodic checkpoint: flush pending messages to session every 5 iterations + // to limit data loss on container crash (#294). Trade-off: partial visibility + // to concurrent reads vs full data loss on crash. + // AddMessage writes to in-memory cache; Save persists to DB. We must clear + // pendingMsgs after AddMessage to prevent double-add in the final flush. + const checkpointInterval = 5 + if iteration > 0 && iteration%checkpointInterval == 0 && len(pendingMsgs) > 0 { + for _, msg := range pendingMsgs { + l.sessions.AddMessage(ctx, req.SessionKey, msg) + } + checkpointFlushedMsgs += len(pendingMsgs) + pendingMsgs = pendingMsgs[:0] + l.sessions.Save(ctx, req.SessionKey) //nolint:errcheck — best-effort persistence + } + } // 4. Full sanitization pipeline (matching TS extractAssistantText + sanitizeUserFacingText) @@ -1451,7 +1467,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // Next time EstimateTokensWithCalibration() is called, it uses this as a base // instead of the chars/3 heuristic (more accurate for multilingual content). if totalUsage.PromptTokens > 0 { - msgCount := len(history) + len(pendingMsgs) + msgCount := len(history) + checkpointFlushedMsgs + len(pendingMsgs) l.sessions.SetLastPromptTokens(ctx, req.SessionKey, totalUsage.PromptTokens, msgCount) } diff --git a/internal/bootstrap/seed_store_test.go b/internal/bootstrap/seed_store_test.go index f260e766..5eb446e0 100644 --- a/internal/bootstrap/seed_store_test.go +++ b/internal/bootstrap/seed_store_test.go @@ -102,6 +102,9 @@ func (s *seedStubStore) UpdateUserProfileMetadata(_ context.Context, _ uuid.UUID func (s *seedStubStore) EnsureUserProfile(_ context.Context, _ uuid.UUID, _ string) error { return nil } +func (s *seedStubStore) PropagateContextFile(_ context.Context, _ uuid.UUID, _ string) (int, error) { + return 0, nil +} // ---- Tests ---- diff --git a/internal/gateway/methods/agents_create_owner_test.go b/internal/gateway/methods/agents_create_owner_test.go index 6e3768aa..0bfb06ed 100644 --- a/internal/gateway/methods/agents_create_owner_test.go +++ b/internal/gateway/methods/agents_create_owner_test.go @@ -95,6 +95,9 @@ func (s *createCaptureStore) UpdateUserProfileMetadata(_ context.Context, _ uuid func (s *createCaptureStore) EnsureUserProfile(_ context.Context, _ uuid.UUID, _ string) error { return nil } +func (s *createCaptureStore) PropagateContextFile(_ context.Context, _ uuid.UUID, _ string) (int, error) { + return 0, nil +} // ---- helpers ---- diff --git a/internal/gateway/methods/agents_files.go b/internal/gateway/methods/agents_files.go index d0dd2ac5..42c390e8 100644 --- a/internal/gateway/methods/agents_files.go +++ b/internal/gateway/methods/agents_files.go @@ -3,6 +3,7 @@ package methods import ( "context" "encoding/json" + "log/slog" "os" "path/filepath" "slices" @@ -215,9 +216,10 @@ func (m *AgentsMethods) handleFilesGet(ctx context.Context, client *gateway.Clie func (m *AgentsMethods) handleFilesSet(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { - AgentID string `json:"agentId"` - Name string `json:"name"` - Content string `json:"content"` + AgentID string `json:"agentId"` + Name string `json:"name"` + Content string `json:"content"` + Propagate bool `json:"propagate"` // push change to all existing user instances } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) @@ -248,6 +250,17 @@ func (m *AgentsMethods) handleFilesSet(ctx context.Context, client *gateway.Clie return } + // Propagate to all existing user instances if requested (#294). + var propagated int + if params.Propagate { + n, err := m.agentStore.PropagateContextFile(ctx, ag.ID, params.Name) + if err != nil { + slog.Warn("agents.files.set: propagation failed", "agent", params.AgentID, "file", params.Name, "error", err) + } else { + propagated = n + } + } + // Invalidate both caches so the new content is served immediately // without waiting for the ContextFileInterceptor's 5-minute TTL to expire. m.agents.InvalidateAgent(params.AgentID) @@ -263,6 +276,7 @@ func (m *AgentsMethods) handleFilesSet(ctx context.Context, client *gateway.Clie "size": len(params.Content), "content": params.Content, }, + "propagated": propagated, })) return } diff --git a/internal/store/agent_store.go b/internal/store/agent_store.go index 3b5e542b..a05c2cb8 100644 --- a/internal/store/agent_store.go +++ b/internal/store/agent_store.go @@ -334,6 +334,10 @@ type AgentStore interface { GetAgentContextFiles(ctx context.Context, agentID uuid.UUID) ([]AgentContextFileData, error) SetAgentContextFile(ctx context.Context, agentID uuid.UUID, fileName, content string) error + // Propagate agent-level file content to all existing user instances that have this file. + // Returns count of updated user rows. + PropagateContextFile(ctx context.Context, agentID uuid.UUID, fileName string) (int, error) + // Per-user context files + overrides GetUserContextFiles(ctx context.Context, agentID uuid.UUID, userID string) ([]UserContextFileData, error) SetUserContextFile(ctx context.Context, agentID uuid.UUID, userID, fileName, content string) error diff --git a/internal/store/pg/agents_context.go b/internal/store/pg/agents_context.go index de4376b0..560e75b3 100644 --- a/internal/store/pg/agents_context.go +++ b/internal/store/pg/agents_context.go @@ -50,6 +50,32 @@ func (s *PGAgentStore) SetAgentContextFile(ctx context.Context, agentID uuid.UUI return err } +// PropagateContextFile copies an agent-level context file to all existing user +// instances that already have that file (seeded users). Returns updated row count. +func (s *PGAgentStore) PropagateContextFile(ctx context.Context, agentID uuid.UUID, fileName string) (int, error) { + tClause, tArgs, err := tenantClauseN(ctx, 4) + if err != nil { + return 0, err + } + // $4 (tenant_id) is referenced twice in the query but only needs one arg value. + res, err := s.db.ExecContext(ctx, + `UPDATE user_context_files + SET content = src.content, updated_at = $3 + FROM ( + SELECT content FROM agent_context_files + WHERE agent_id = $1 AND file_name = $2`+tClause+` + ) src + WHERE user_context_files.agent_id = $1 + AND user_context_files.file_name = $2`+tClause, + append([]any{agentID, fileName, time.Now()}, tArgs...)..., + ) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + return int(n), nil +} + // --- Per-user Context Files --- func (s *PGAgentStore) GetUserContextFiles(ctx context.Context, agentID uuid.UUID, userID string) ([]store.UserContextFileData, error) { diff --git a/internal/tools/context_file_interceptor_test.go b/internal/tools/context_file_interceptor_test.go index 3c37c0a2..63d059e9 100644 --- a/internal/tools/context_file_interceptor_test.go +++ b/internal/tools/context_file_interceptor_test.go @@ -84,6 +84,9 @@ func (s *stubAgentStore) UpdateUserProfileMetadata(_ context.Context, _ uuid.UUI func (s *stubAgentStore) EnsureUserProfile(_ context.Context, _ uuid.UUID, _ string) error { return nil } +func (s *stubAgentStore) PropagateContextFile(_ context.Context, _ uuid.UUID, _ string) (int, error) { + return 0, nil +} // ---- Tests ----