Unified Agent + reusable Go agent core.
feat: persist heartbeat memory updates and simplify flow
+197
-44
+34
-15
internal/channelruntime/telegram/memory_flow.go
+34
-15
internal/channelruntime/telegram/memory_flow.go
···
16
16
"github.com/quailyquaily/mistermorph/memory"
17
17
)
18
18
19
-
func updateTelegramMemory(ctx context.Context, logger *slog.Logger, client llm.Client, model string, mgr *memory.Manager, id memory.Identity, job telegramJob, history []chathistory.ChatHistoryItem, historyCap int, final *agent.Final, requestTimeout time.Duration) error {
19
+
func updateMemoryFromJob(ctx context.Context, logger *slog.Logger, client llm.Client, model string, mgr *memory.Manager, longTermSubjectID string, job telegramJob, history []chathistory.ChatHistoryItem, historyCap int, final *agent.Final, requestTimeout time.Duration) error {
20
20
if mgr == nil || client == nil {
21
21
return nil
22
22
}
23
23
output := formatFinalOutput(final)
24
24
date := time.Now().UTC()
25
-
contactNickname := strings.TrimSpace(job.FromDisplayName)
26
-
if contactNickname == "" {
27
-
contactNickname = strings.TrimSpace(strings.Join([]string{job.FromFirstName, job.FromLastName}, " "))
28
-
}
29
-
meta := memory.WriteMeta{
30
-
SessionID: fmt.Sprintf("tg:%d", job.ChatID),
31
-
ContactIDs: []string{telegramMemoryContactID(job.FromUsername, job.FromUserID)},
32
-
ContactNicknames: []string{contactNickname},
33
-
}
25
+
longTermSubjectID = strings.TrimSpace(longTermSubjectID)
26
+
meta := buildMemoryWriteMeta(job)
34
27
35
28
ctxInfo := MemoryDraftContext{
36
29
SessionID: meta.SessionID,
···
57
50
defer cancel()
58
51
ctxInfo.CounterpartyLabel = buildMemoryCounterpartyLabel(meta, ctxInfo)
59
52
60
-
draftHistory := buildTelegramMemoryDraftHistory(history, job, output, date, historyCap)
53
+
draftHistory := buildMemoryDraftHistory(history, job, output, date, historyCap)
61
54
draft, err := BuildMemoryDraft(memCtx, client, model, draftHistory, job.Text, output, existingContent, ctxInfo)
62
55
if err != nil {
63
56
return err
···
80
73
if err != nil {
81
74
return err
82
75
}
83
-
if _, err := mgr.UpdateLongTerm(id.SubjectID, draft.Promote); err != nil {
84
-
return err
76
+
if longTermSubjectID != "" {
77
+
if _, err := mgr.UpdateLongTerm(longTermSubjectID, draft.Promote); err != nil {
78
+
return err
79
+
}
85
80
}
86
81
if logger != nil {
87
-
logger.Debug("memory_update_ok", "subject_id", id.SubjectID)
82
+
attrs := []any{"session_id", meta.SessionID}
83
+
if longTermSubjectID != "" {
84
+
attrs = append(attrs, "subject_id", longTermSubjectID)
85
+
}
86
+
logger.Debug("memory_update_ok", attrs...)
88
87
}
89
88
return nil
90
89
}
91
90
92
-
func buildTelegramMemoryDraftHistory(history []chathistory.ChatHistoryItem, job telegramJob, output string, now time.Time, maxItems int) []chathistory.ChatHistoryItem {
91
+
const heartbeatMemorySessionID = "heartbeat"
92
+
93
+
func buildMemoryWriteMeta(job telegramJob) memory.WriteMeta {
94
+
if job.IsHeartbeat {
95
+
return memory.WriteMeta{SessionID: heartbeatMemorySessionID}
96
+
}
97
+
meta := memory.WriteMeta{SessionID: fmt.Sprintf("tg:%d", job.ChatID)}
98
+
if contactID := telegramMemoryContactID(job.FromUsername, job.FromUserID); contactID != "" {
99
+
meta.ContactIDs = []string{contactID}
100
+
}
101
+
contactNickname := strings.TrimSpace(job.FromDisplayName)
102
+
if contactNickname == "" {
103
+
contactNickname = strings.TrimSpace(strings.Join([]string{job.FromFirstName, job.FromLastName}, " "))
104
+
}
105
+
if contactNickname != "" {
106
+
meta.ContactNicknames = []string{contactNickname}
107
+
}
108
+
return meta
109
+
}
110
+
111
+
func buildMemoryDraftHistory(history []chathistory.ChatHistoryItem, job telegramJob, output string, now time.Time, maxItems int) []chathistory.ChatHistoryItem {
93
112
out := append([]chathistory.ChatHistoryItem{}, history...)
94
113
out = append(out, newTelegramInboundHistoryItem(job))
95
114
if strings.TrimSpace(output) != "" {
+43
internal/channelruntime/telegram/memory_flow_test.go
+43
internal/channelruntime/telegram/memory_flow_test.go
···
1
+
package telegram
2
+
3
+
import (
4
+
"testing"
5
+
)
6
+
7
+
func TestBuildMemoryWriteMeta(t *testing.T) {
8
+
t.Run("heartbeat uses fixed session path", func(t *testing.T) {
9
+
meta := buildMemoryWriteMeta(telegramJob{
10
+
IsHeartbeat: true,
11
+
ChatID: 42,
12
+
FromUserID: 1001,
13
+
FromUsername: "alice",
14
+
})
15
+
if meta.SessionID != heartbeatMemorySessionID {
16
+
t.Fatalf("session_id = %q, want %q", meta.SessionID, heartbeatMemorySessionID)
17
+
}
18
+
if len(meta.ContactIDs) != 0 {
19
+
t.Fatalf("contact_ids = %#v, want empty", meta.ContactIDs)
20
+
}
21
+
if len(meta.ContactNicknames) != 0 {
22
+
t.Fatalf("contact_nicknames = %#v, want empty", meta.ContactNicknames)
23
+
}
24
+
})
25
+
26
+
t.Run("normal chat keeps tg session and contact meta", func(t *testing.T) {
27
+
meta := buildMemoryWriteMeta(telegramJob{
28
+
ChatID: 777,
29
+
FromUserID: 1001,
30
+
FromUsername: "@alice",
31
+
FromDisplayName: "Alice",
32
+
})
33
+
if meta.SessionID != "tg:777" {
34
+
t.Fatalf("session_id = %q, want %q", meta.SessionID, "tg:777")
35
+
}
36
+
if len(meta.ContactIDs) != 1 || meta.ContactIDs[0] != "tg:@alice" {
37
+
t.Fatalf("contact_ids = %#v, want [\"tg:@alice\"]", meta.ContactIDs)
38
+
}
39
+
if len(meta.ContactNicknames) != 1 || meta.ContactNicknames[0] != "Alice" {
40
+
t.Fatalf("contact_nicknames = %#v, want [\"Alice\"]", meta.ContactNicknames)
41
+
}
42
+
})
43
+
}
+51
-29
internal/channelruntime/telegram/runtime_task.go
+51
-29
internal/channelruntime/telegram/runtime_task.go
···
91
91
92
92
var memManager *memory.Manager
93
93
var memIdentity memory.Identity
94
-
memReqCtx := memory.ContextPublic
95
-
if runtimeOpts.MemoryEnabled && job.FromUserID > 0 {
96
-
if strings.ToLower(strings.TrimSpace(job.ChatType)) == "private" {
97
-
memReqCtx = memory.ContextPrivate
98
-
}
99
-
id, err := (&memory.Resolver{}).ResolveTelegram(ctx, job.FromUserID)
100
-
if err != nil {
101
-
return nil, nil, loadedSkills, nil, fmt.Errorf("memory identity: %w", err)
102
-
}
103
-
if id.Enabled && strings.TrimSpace(id.SubjectID) != "" {
104
-
memIdentity = id
105
-
memManager = memory.NewManager(statepaths.MemoryDir(), runtimeOpts.MemoryShortTermDays)
106
-
if runtimeOpts.MemoryInjectionEnabled {
107
-
maxItems := runtimeOpts.MemoryInjectionMaxItems
108
-
snap, err := memManager.BuildInjection(id.SubjectID, memReqCtx, maxItems)
109
-
if err != nil {
110
-
return nil, nil, loadedSkills, nil, fmt.Errorf("memory injection: %w", err)
111
-
}
112
-
if strings.TrimSpace(snap) != "" {
113
-
promptprofile.AppendMemorySummariesBlock(&promptSpec, snap)
114
-
if logger != nil {
115
-
logger.Info("memory_injection_applied", "source", "telegram", "subject_id", id.SubjectID, "chat_id", job.ChatID, "snapshot_len", len(snap))
94
+
if runtimeOpts.MemoryEnabled {
95
+
memManager = memory.NewManager(statepaths.MemoryDir(), runtimeOpts.MemoryShortTermDays)
96
+
if job.FromUserID > 0 {
97
+
memReqCtx := memory.ContextPublic
98
+
if strings.ToLower(strings.TrimSpace(job.ChatType)) == "private" {
99
+
memReqCtx = memory.ContextPrivate
100
+
}
101
+
id, err := (&memory.Resolver{}).ResolveTelegram(ctx, job.FromUserID)
102
+
if err != nil {
103
+
return nil, nil, loadedSkills, nil, fmt.Errorf("memory identity: %w", err)
104
+
}
105
+
if id.Enabled && strings.TrimSpace(id.SubjectID) != "" {
106
+
memIdentity = id
107
+
if runtimeOpts.MemoryInjectionEnabled {
108
+
maxItems := runtimeOpts.MemoryInjectionMaxItems
109
+
snap, err := memManager.BuildInjection(id.SubjectID, memReqCtx, maxItems)
110
+
if err != nil {
111
+
return nil, nil, loadedSkills, nil, fmt.Errorf("memory injection: %w", err)
112
+
}
113
+
if strings.TrimSpace(snap) != "" {
114
+
promptprofile.AppendMemorySummariesBlock(&promptSpec, snap)
115
+
if logger != nil {
116
+
logger.Info("memory_injection_applied", "source", "telegram", "subject_id", id.SubjectID, "chat_id", job.ChatID, "snapshot_len", len(snap))
117
+
}
118
+
} else if logger != nil {
119
+
logger.Debug("memory_injection_skipped", "source", "telegram", "reason", "empty_snapshot", "subject_id", id.SubjectID, "chat_id", job.ChatID)
116
120
}
117
121
} else if logger != nil {
118
-
logger.Debug("memory_injection_skipped", "source", "telegram", "reason", "empty_snapshot", "subject_id", id.SubjectID, "chat_id", job.ChatID)
122
+
logger.Debug("memory_injection_skipped", "source", "telegram", "reason", "disabled")
119
123
}
120
124
} else if logger != nil {
121
-
logger.Debug("memory_injection_skipped", "source", "telegram", "reason", "disabled")
125
+
logger.Debug("memory_identity_unavailable", "source", "telegram", "enabled", id.Enabled, "subject_id", strings.TrimSpace(id.SubjectID))
122
126
}
123
-
} else if logger != nil {
124
-
logger.Debug("memory_identity_unavailable", "source", "telegram", "enabled", id.Enabled, "subject_id", strings.TrimSpace(id.SubjectID))
127
+
} else if logger != nil && !job.IsHeartbeat {
128
+
logger.Debug("memory_identity_unavailable", "source", "telegram", "reason", "missing_user_id")
125
129
}
126
130
}
127
131
···
201
205
202
206
publishText := shouldPublishTelegramText(final)
203
207
204
-
if publishText && !job.IsHeartbeat && memManager != nil && memIdentity.Enabled && strings.TrimSpace(memIdentity.SubjectID) != "" {
205
-
if err := updateTelegramMemory(ctx, logger, client, model, memManager, memIdentity, job, history, historyCap, final, requestTimeout); err != nil {
208
+
longTermSubjectID := resolveLongTermSubjectID(job, memIdentity)
209
+
if shouldWriteMemory(publishText, memManager, longTermSubjectID) {
210
+
if err := updateMemoryFromJob(ctx, logger, client, model, memManager, longTermSubjectID, job, history, historyCap, final, requestTimeout); err != nil {
206
211
if errors.Is(err, context.DeadlineExceeded) {
207
212
retryutil.AsyncRetry(logger, "memory_update", 2*time.Second, requestTimeout, func(retryCtx context.Context) error {
208
-
return updateTelegramMemory(retryCtx, logger, client, model, memManager, memIdentity, job, history, historyCap, final, requestTimeout)
213
+
return updateMemoryFromJob(retryCtx, logger, client, model, memManager, longTermSubjectID, job, history, historyCap, final, requestTimeout)
209
214
})
210
215
}
211
216
logger.Warn("memory_update_error", "error", err.Error())
212
217
}
213
218
}
214
219
return final, agentCtx, loadedSkills, reaction, nil
220
+
}
221
+
222
+
func resolveLongTermSubjectID(job telegramJob, memIdentity memory.Identity) string {
223
+
if job.IsHeartbeat {
224
+
return heartbeatMemorySessionID
225
+
}
226
+
if !memIdentity.Enabled {
227
+
return ""
228
+
}
229
+
return strings.TrimSpace(memIdentity.SubjectID)
230
+
}
231
+
232
+
func shouldWriteMemory(publishText bool, memManager *memory.Manager, longTermSubjectID string) bool {
233
+
if !publishText || memManager == nil {
234
+
return false
235
+
}
236
+
return strings.TrimSpace(longTermSubjectID) != ""
215
237
}
216
238
217
239
func runMAEPTask(ctx context.Context, d Dependencies, logger *slog.Logger, logOpts agent.LogOptions, client llm.Client, baseReg *tools.Registry, sharedGuard *guard.Guard, cfg agent.Config, model string, peerID string, history []llm.Message, stickySkills []string, runtimeOpts runtimeTaskOptions, task string) (*agent.Final, *agent.Context, []string, error) {
+69
internal/channelruntime/telegram/runtime_task_test.go
+69
internal/channelruntime/telegram/runtime_task_test.go
···
1
+
package telegram
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"github.com/quailyquaily/mistermorph/memory"
7
+
)
8
+
9
+
func TestShouldWriteMemory(t *testing.T) {
10
+
mgr := &memory.Manager{}
11
+
12
+
tests := []struct {
13
+
name string
14
+
publishText bool
15
+
memManager *memory.Manager
16
+
longTermSubjectID string
17
+
want bool
18
+
}{
19
+
{
20
+
name: "skip when output is not published",
21
+
publishText: false,
22
+
memManager: mgr,
23
+
longTermSubjectID: heartbeatMemorySessionID,
24
+
want: false,
25
+
},
26
+
{
27
+
name: "skip when memory manager is missing",
28
+
publishText: true,
29
+
memManager: nil,
30
+
longTermSubjectID: heartbeatMemorySessionID,
31
+
want: false,
32
+
},
33
+
{
34
+
name: "write when long-term subject is resolved",
35
+
publishText: true,
36
+
memManager: mgr,
37
+
longTermSubjectID: heartbeatMemorySessionID,
38
+
want: true,
39
+
},
40
+
{
41
+
name: "skip when long-term subject is empty",
42
+
publishText: true,
43
+
memManager: mgr,
44
+
longTermSubjectID: "",
45
+
want: false,
46
+
},
47
+
}
48
+
49
+
for _, tc := range tests {
50
+
t.Run(tc.name, func(t *testing.T) {
51
+
got := shouldWriteMemory(tc.publishText, tc.memManager, tc.longTermSubjectID)
52
+
if got != tc.want {
53
+
t.Fatalf("shouldWriteMemory() = %v, want %v", got, tc.want)
54
+
}
55
+
})
56
+
}
57
+
}
58
+
59
+
func TestResolveLongTermSubjectID(t *testing.T) {
60
+
if got := resolveLongTermSubjectID(telegramJob{IsHeartbeat: true}, memory.Identity{}); got != heartbeatMemorySessionID {
61
+
t.Fatalf("heartbeat subject = %q, want %q", got, heartbeatMemorySessionID)
62
+
}
63
+
if got := resolveLongTermSubjectID(telegramJob{}, memory.Identity{Enabled: true, SubjectID: "ext:telegram:1"}); got != "ext:telegram:1" {
64
+
t.Fatalf("normal subject = %q, want %q", got, "ext:telegram:1")
65
+
}
66
+
if got := resolveLongTermSubjectID(telegramJob{}, memory.Identity{Enabled: false, SubjectID: "ext:telegram:1"}); got != "" {
67
+
t.Fatalf("disabled identity subject = %q, want empty", got)
68
+
}
69
+
}