fix: address 4 bugs from issue #294 — context propagation, cron reliability, crash safety

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

@@ -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))
 	})

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{

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)
 	}
 

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 ----
 

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 ----
 

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, &params)
@@ -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
 	}

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

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) {

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 ----