Unified Agent + reusable Go agent core.
fix: keep console store unchanged on reload failure
+118
-12
+32
-12
internal/daemonruntime/console_store.go
+32
-12
internal/daemonruntime/console_store.go
···
101
101
heartbeatTopicID = "_heartbeat"
102
102
}
103
103
now := time.Now().UTC()
104
+
nextRootDir := filepath.Clean(rootDir)
105
+
nextLogDir := filepath.Join(nextRootDir, "log")
106
+
nextTopicPath := filepath.Join(nextRootDir, "topic.json")
104
107
105
108
s.mu.Lock()
106
109
defer s.mu.Unlock()
···
108
111
oldRootDir := s.rootDir
109
112
oldPersist := s.persist
110
113
111
-
s.rootDir = filepath.Clean(rootDir)
112
-
s.logDir = filepath.Join(filepath.Clean(rootDir), "log")
113
-
s.topicPath = filepath.Join(filepath.Clean(rootDir), "topic.json")
114
-
s.heartbeatTopicID = heartbeatTopicID
115
-
s.persist = opts.Persist
116
-
117
-
if !s.persist {
114
+
if !opts.Persist {
115
+
s.rootDir = nextRootDir
116
+
s.logDir = nextLogDir
117
+
s.topicPath = nextTopicPath
118
+
s.heartbeatTopicID = heartbeatTopicID
119
+
s.persist = false
118
120
return nil
119
121
}
120
-
if err := s.persistTopicsLocked(now); err != nil {
122
+
if err := s.persistTopicsAtPathLocked(nextTopicPath, now); err != nil {
121
123
return err
122
124
}
123
-
if oldPersist && s.rootDir == oldRootDir {
125
+
if oldPersist && nextRootDir == oldRootDir {
126
+
s.rootDir = nextRootDir
127
+
s.logDir = nextLogDir
128
+
s.topicPath = nextTopicPath
129
+
s.heartbeatTopicID = heartbeatTopicID
130
+
s.persist = true
124
131
return nil
125
132
}
126
133
for _, item := range s.items {
127
-
if err := s.appendTaskEventLocked(item, now, s.triggerForTaskLocked(item.ID, TaskTrigger{})); err != nil {
134
+
if err := s.appendTaskEventAtLogDirLocked(nextLogDir, item, now, s.triggerForTaskLocked(item.ID, TaskTrigger{})); err != nil {
128
135
return err
129
136
}
130
137
}
138
+
s.rootDir = nextRootDir
139
+
s.logDir = nextLogDir
140
+
s.topicPath = nextTopicPath
141
+
s.heartbeatTopicID = heartbeatTopicID
142
+
s.persist = true
131
143
return nil
132
144
}
133
145
···
541
553
if !s.persist {
542
554
return nil
543
555
}
544
-
path := filepath.Join(s.logDir, fmt.Sprintf("%s_%s.jsonl", now.Format("2006-01-02"), consoleTopicKey(info.TopicID)))
556
+
return s.appendTaskEventAtLogDirLocked(s.logDir, info, now, trigger)
557
+
}
558
+
559
+
func (s *ConsoleFileStore) appendTaskEventAtLogDirLocked(logDir string, info TaskInfo, now time.Time, trigger TaskTrigger) error {
560
+
path := filepath.Join(logDir, fmt.Sprintf("%s_%s.jsonl", now.Format("2006-01-02"), consoleTopicKey(info.TopicID)))
545
561
writer, err := fsstore.NewJSONLWriter(path, fsstore.JSONLOptions{
546
562
RotateMaxBytes: 1 << 60,
547
563
SyncEachWrite: true,
···
568
584
if !s.persist {
569
585
return nil
570
586
}
587
+
return s.persistTopicsAtPathLocked(s.topicPath, now)
588
+
}
589
+
590
+
func (s *ConsoleFileStore) persistTopicsAtPathLocked(topicPath string, now time.Time) error {
571
591
topics := make([]TopicInfo, 0, len(s.topics))
572
592
for _, topic := range s.topics {
573
593
topics = append(topics, normalizeTopicInfo(topic))
···
578
598
}
579
599
return topics[i].CreatedAt.Before(topics[j].CreatedAt)
580
600
})
581
-
return fsstore.WriteJSONAtomic(s.topicPath, consoleTopicFile{
601
+
return fsstore.WriteJSONAtomic(topicPath, consoleTopicFile{
582
602
Version: consoleTopicFileVersion,
583
603
UpdatedAt: now,
584
604
Items: topics,
+86
internal/daemonruntime/console_store_test.go
+86
internal/daemonruntime/console_store_test.go
···
1
1
package daemonruntime
2
2
3
3
import (
4
+
"os"
5
+
"path/filepath"
4
6
"testing"
5
7
"time"
6
8
)
···
195
197
topics := store.ListTopics()
196
198
if len(topics) != 0 {
197
199
t.Fatalf("len(topics) = %d, want 0", len(topics))
200
+
}
201
+
}
202
+
203
+
func TestConsoleFileStoreApplyConfigDoesNotMutateStateOnRewriteFailure(t *testing.T) {
204
+
oldRoot := t.TempDir()
205
+
store, err := NewConsoleFileStore(ConsoleFileStoreOptions{
206
+
RootDir: oldRoot,
207
+
HeartbeatTopicID: "_heartbeat",
208
+
Persist: true,
209
+
})
210
+
if err != nil {
211
+
t.Fatalf("NewConsoleFileStore() error = %v", err)
212
+
}
213
+
214
+
if err := store.UpsertWithTrigger(TaskInfo{
215
+
ID: "task_before_apply_config_failure",
216
+
Status: TaskQueued,
217
+
Task: "hello",
218
+
Model: "gpt-5.2",
219
+
Timeout: "10m0s",
220
+
CreatedAt: mustParseTime(t, "2026-03-15T10:03:00Z"),
221
+
TopicID: ConsoleDefaultTopicID,
222
+
}, TaskTrigger{Source: "ui", Event: "chat_submit"}, ""); err != nil {
223
+
t.Fatalf("UpsertWithTrigger() error = %v", err)
224
+
}
225
+
226
+
oldLogDir := store.logDir
227
+
oldTopicPath := store.topicPath
228
+
oldHeartbeatTopicID := store.heartbeatTopicID
229
+
oldPersist := store.persist
230
+
231
+
nextRoot := t.TempDir()
232
+
blockedLogPath := filepath.Join(nextRoot, "log")
233
+
if err := os.WriteFile(blockedLogPath, []byte("not-a-dir"), 0o600); err != nil {
234
+
t.Fatalf("WriteFile(%q) error = %v", blockedLogPath, err)
235
+
}
236
+
237
+
err = store.ApplyConfig(ConsoleFileStoreOptions{
238
+
RootDir: nextRoot,
239
+
HeartbeatTopicID: "_heartbeat_next",
240
+
Persist: true,
241
+
})
242
+
if err == nil {
243
+
t.Fatal("ApplyConfig() error = nil, want rewrite failure")
244
+
}
245
+
246
+
if store.rootDir != oldRoot {
247
+
t.Fatalf("store.rootDir = %q, want %q", store.rootDir, oldRoot)
248
+
}
249
+
if store.logDir != oldLogDir {
250
+
t.Fatalf("store.logDir = %q, want %q", store.logDir, oldLogDir)
251
+
}
252
+
if store.topicPath != oldTopicPath {
253
+
t.Fatalf("store.topicPath = %q, want %q", store.topicPath, oldTopicPath)
254
+
}
255
+
if store.heartbeatTopicID != oldHeartbeatTopicID {
256
+
t.Fatalf("store.heartbeatTopicID = %q, want %q", store.heartbeatTopicID, oldHeartbeatTopicID)
257
+
}
258
+
if store.persist != oldPersist {
259
+
t.Fatalf("store.persist = %v, want %v", store.persist, oldPersist)
260
+
}
261
+
262
+
if err := store.UpsertWithTrigger(TaskInfo{
263
+
ID: "task_after_apply_config_failure",
264
+
Status: TaskDone,
265
+
Task: "still old root",
266
+
Model: "gpt-5.2",
267
+
Timeout: "10m0s",
268
+
CreatedAt: mustParseTime(t, "2026-03-15T10:04:00Z"),
269
+
TopicID: ConsoleDefaultTopicID,
270
+
}, TaskTrigger{Source: "ui", Event: "chat_submit"}, ""); err != nil {
271
+
t.Fatalf("UpsertWithTrigger(after failure) error = %v", err)
272
+
}
273
+
274
+
reloaded, err := NewConsoleFileStore(ConsoleFileStoreOptions{
275
+
RootDir: oldRoot,
276
+
HeartbeatTopicID: "_heartbeat",
277
+
Persist: true,
278
+
})
279
+
if err != nil {
280
+
t.Fatalf("reload NewConsoleFileStore() error = %v", err)
281
+
}
282
+
if _, ok := reloaded.Get("task_after_apply_config_failure"); !ok {
283
+
t.Fatal("task_after_apply_config_failure missing from old root after failed ApplyConfig")
198
284
}
199
285
}
200
286