From 2283383b9e371cd4dff12419177611bfc9226d55 Mon Sep 17 00:00:00 2001 From: Aska Date: Mon, 7 Sep 2026 19:34:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(backthink):=20=E8=87=AA=E4=B8=BB=E6=80=9D?= =?UTF-8?q?=E8=80=83=20P2=E2=80=94=E2=80=94=E7=BB=93=E6=9E=84=E5=8C=96?= =?UTF-8?q?=E6=8E=A8=E9=80=81=E5=86=B3=E7=AD=96+=E6=B8=A0=E9=81=93?= =?UTF-8?q?=E9=80=89=E6=8B=A9(design/20)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - proactive.go(新): parseProactive 从结论解析 [我决定:push|no_push][渠道:...]<正文>; 兼容大小写/全半角冒号; 无标记回退(无/nothing=不推); 畸形/孤标记 fail-safe 不推 - context.go: guidancePrompt 引导结构化结尾+保留P1温暖引导 - push.go: PushRequest{Message,ChannelHint}; Pusher.Push 路由(auto→active优先/IM兜底, web→仅active, im→仅IM, 具体平台→IM+platform打标+缺IM回退默认记detail); buildProactiveMessage - backthink.go: think() 用 parseProactive→ShouldPush=push&&disturbAllowed(不再被tr.Import卡死) - decide.go: disturbAllowed(asleep/working/quiet 硬性不打扰门) 取代旧import门(保留旧Decide兼容) - 测试: parseProactive 6组+think端到端(真实pusher:kook→IM/无标记→active/no_push→不推/睡眠→门拦截) +buildProactiveMessage+具体平台无IM回退; 后端 build/vet/test 全绿 - 注: 生产 Active/IM sink 未注入 wire.go backthink.Deps(另一步); 多平台IM受单IMAdapter接口限制 --- internal/services/backthink/backthink.go | 35 +- internal/services/backthink/context.go | 10 + internal/services/backthink/decide.go | 28 ++ internal/services/backthink/mocks_test.go | 14 +- internal/services/backthink/proactive.go | 154 +++++++++ internal/services/backthink/proactive_test.go | 309 ++++++++++++++++++ internal/services/backthink/push.go | 167 ++++++++-- internal/services/backthink/push_test.go | 18 +- 8 files changed, 688 insertions(+), 47 deletions(-) create mode 100644 internal/services/backthink/proactive.go create mode 100644 internal/services/backthink/proactive_test.go diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index f46b8df..046691f 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -390,26 +390,41 @@ func (b *Backthink) think(parent context.Context, tr Trigger) { return } - state := b.currentState(cctx) - decision := b.Decide(DecisionInput{Trigger: tr.Source, Import: tr.Import, At: at, UserState: state}) + // P2 (design/20 §2): her own conclusion decides whether to push and to which + // channel — not the trigger's importance. parseProactive recovers the decision + // robustly and fails safe (malformed/ambiguous → no push). + message, channelHint, push := parseProactive(conclusion) - pushRes := PushResult{Route: RouteNone, Detail: decision.Reason} - if decision.ShouldPush { - pushRes = b.pusher.Push(cctx, b.buildProactiveMessage(conclusion)) + state := b.currentState(cctx) + // The "do not disturb" gate (asleep/working/busy/quiet) still applies as a hard + // safety; the frequency gate (PushGate) is applied inside the pusher. + shouldPush := push && b.disturbAllowed(at, state) + + recordConclusion := message + if recordConclusion == "" { + recordConclusion = conclusion + } + + pushRes := PushResult{Route: RouteNone, Detail: b.interruptReason(push, at, state)} + if shouldPush { + pushRes = b.pusher.Push(cctx, PushRequest{ + Message: b.buildProactiveMessage(message, channelHint), + ChannelHint: channelHint, + }) } b.record(Record{ Trigger: tr.Source, At: at, - Conclusion: conclusion, - ShouldPush: decision.ShouldPush, + Conclusion: recordConclusion, + ShouldPush: shouldPush, PushResult: pushRes, }) - if decision.ShouldPush { - b.log.Info("backthink proactive push", "source", tr.Source, "route", pushRes.Route, "pushed", pushRes.Pushed, "reason", decision.Reason) + if shouldPush { + b.log.Info("backthink proactive push", "source", tr.Source, "route", pushRes.Route, "pushed", pushRes.Pushed, "reason", pushRes.Detail) } else { - b.log.Debug("backthink did not interrupt", "source", tr.Source, "reason", decision.Reason) + b.log.Debug("backthink did not interrupt", "source", tr.Source, "reason", pushRes.Detail) } } diff --git a/internal/services/backthink/context.go b/internal/services/backthink/context.go index 06ed015..c302fa7 100644 --- a/internal/services/backthink/context.go +++ b/internal/services/backthink/context.go @@ -84,6 +84,12 @@ func thinkInput(recent []model.Message, trigger string) []model.Message { // never on this prompt text (red line); the prompt only describes the running // environment (memories readable/writable, tools callable) that the env already // grants. +// +// The prompt asks for a compact structured closing — a push/no_push decision, an +// optional target channel and the message body — so the service can honor her own +// judgment instead of gating on the trigger's importance (design/20 §2 P2). The +// format is guidance, not a contract: parseProactive fails safe when she omits or +// mangles it. func guidancePrompt(trigger string) string { return "[自主思考 · 无人在旁]\n" + "触发：" + trigger + "\n" + @@ -91,5 +97,9 @@ func guidancePrompt(trigger string) string { "请回顾最近与主人的相处、她的状态、你们共同经历的事与临近的日程，判断此刻是否有一件真正值得主动告诉她的、温暖而有针对性的事情——" + "比如关心她的近况、记起她曾说过的某件事、提醒一件即将到来的日程、见证一个属于你们的共同时刻，或注意到某个值得留意的变化。\n" + "若有，就用温暖自然的语气，用一到两句把它说出来——这正是你此刻最想对她说的话；若确实没有值得说的，只需回复「无」，也不要编造。\n" + + "请把「是否主动说」与「发到哪个渠道」写成一个简短、格式固定的结尾，解析规则如下：\n" + + "[我决定:push] 或 [我决定:no_push]\n" + + "[渠道:web|im|kook|telegram|...|auto]（可省略，auto/省略即交给默认：优先网页端、IM 兜底）\n" + + "上面两行之后，紧接你想说的话作为正文。\n" + "整轮思考保持简短、单轮，不反复自我触发。" } diff --git a/internal/services/backthink/decide.go b/internal/services/backthink/decide.go index 26ac40c..bd4f2a2 100644 --- a/internal/services/backthink/decide.go +++ b/internal/services/backthink/decide.go @@ -128,6 +128,34 @@ func (b *Backthink) Decide(in DecisionInput) Decision { } } +// disturbAllowed is the hard "do not disturb" safety gate for an autonomous push +// (design/12 §2 first principle). Unlike Decide it deliberately ignores trigger +// importance — which design/20 P2 replaces with her own judgment — and only blocks +// when it is genuinely inappropriate to interrupt: the user is asleep, working or +// busy, or it is inside the quiet-hours window. The frequency gate (PushGate) +// applies separately inside the pusher. +func (b *Backthink) disturbAllowed(at time.Time, s UserState) bool { + if s.Sleeping || s.Working || s.Busy { + return false + } + return !b.inQuietHours(at) +} + +// interruptReason is the short human-readable note recorded/logged when no push +// goes out (or as the pre-push intent). It keeps the panel self-explanatory: +// either she concluded no push, or the do-not-disturb gate refused a push she +// wanted. +func (b *Backthink) interruptReason(push bool, at time.Time, s UserState) string { + switch { + case !push: + return "conclusion requested no push" + case !b.disturbAllowed(at, s): + return "proactive push blocked by do-not-disturb" + default: + return "proactive push approved" + } +} + // inQuietHours reports whether at falls inside the configured do-not-disturb // window. The window may wrap midnight (e.g. 23:00-07:00); two equal non-zero // hours disable it. diff --git a/internal/services/backthink/mocks_test.go b/internal/services/backthink/mocks_test.go index b65fbdb..c283704 100644 --- a/internal/services/backthink/mocks_test.go +++ b/internal/services/backthink/mocks_test.go @@ -52,16 +52,18 @@ func (m *mockState) Current(context.Context) (UserState, error) { return m.state // mockPusher records Push calls and returns a fixed result. type mockPusher struct { - mu sync.Mutex - result PushResult - calls int - lastMsg adapter.UnifiedMessage + mu sync.Mutex + result PushResult + calls int + lastMsg adapter.UnifiedMessage + lastHint string } -func (p *mockPusher) Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult { +func (p *mockPusher) Push(ctx context.Context, req PushRequest) PushResult { p.mu.Lock() p.calls++ - p.lastMsg = msg + p.lastMsg = req.Message + p.lastHint = req.ChannelHint p.mu.Unlock() return p.result } diff --git a/internal/services/backthink/proactive.go b/internal/services/backthink/proactive.go new file mode 100644 index 0000000..238267f --- /dev/null +++ b/internal/services/backthink/proactive.go @@ -0,0 +1,154 @@ +package backthink + +import ( + "regexp" + "strings" + "unicode" +) + +// The autonomous think is guided (guidancePrompt in context.go) to conclude with +// a compact, machine-parseable decision so the service no longer has to infer +// whether to push from the trigger's importance alone (design/20 §2 P2): +// +// [我决定:push] (or no_push) +// [渠道:web|im|kook|telegram|...|auto] +// +// +// The markers are parsed as plain text so a model that formats them slightly +// differently still yields a usable decision; a malformed/ambiguous decision +// marker fails safe to "no push". +var ( + // decideMarker matches the push/no_push decision token, tolerating a colon in + // either width and optional spaces. + decideMarker = regexp.MustCompile(`(?i)\[\s*我决定\s*[:：]\s*(no_push|push)\s*\]`) + // channelMarker matches the target-channel token. + channelMarker = regexp.MustCompile(`(?i)\[\s*渠道\s*[:：]\s*([a-z0-9_\-]+)\s*\]`) + // decidePrefixes are the marker shapes that signal a decision was attempted + // but could not be parsed, so the pass is treated as ambiguous (fail safe). It + // is deliberately the bracketed form only, so a message that merely contains + // "我决定：" in prose (common in Chinese) is not misread as a malformed marker. + decidePrefixes = []string{"[我决定"} +) + +// nonePhrases are conclusions that mean "nothing to report" when no marker is +// present. The body is normalized (case-folded, punctuation/spaces stripped) +// before comparison so "无", "无。" and "Nothing to report." all collapse to the +// same token. +var nonePhrases = []string{ + "无", "没什么", "没有", "无物可报", "暂无", "没有值得说的", + "nothing", "none", "nothingtoreport", "nothingworthreporting", "nothingnew", +} + +// parseProactive recovers the push decision, the target-channel hint and the +// clean message body from a think conclusion. It is a pure function: it reads no +// state and performs no I/O, so it is safe to call concurrently and cheap to test. +// +// Semantics (design/20 §2 P2): +// - A well-formed [我决定:push|no_push] marker is authoritative. +// - A malformed/ambiguous decision marker → no push (fail safe). +// - No marker: a body that reads as "无"/"nothing to report" → no push; +// otherwise the body is treated as a proactive message (push=true), matching +// the pre-P2 default. +// - The channel hint is the [渠道:...] token, lowercased; empty/auto keeps the +// default active-client-first, IM-fallback route. +// - A push decision whose body is empty or still reads as "nothing" → no push +// (fail safe), so a lone marker never pushes an empty message. +func parseProactive(conclusion string) (message, channelHint string, push bool) { + text := strings.TrimSpace(conclusion) + + if val, ok := matchDecide(text); ok { + push = val == "push" + } else if hasAmbiguousDecide(text) { + push = false + } else { + push = !looksNone(text) + } + + channelHint = matchChannel(text) + message = stripMarkers(text) + + if push && (message == "" || looksNone(message)) { + push = false + } + return message, channelHint, push +} + +// matchDecide returns the matched decision token (lowercased) and whether a +// well-formed decision marker was present. +func matchDecide(text string) (string, bool) { + m := decideMarker.FindStringSubmatch(text) + if m == nil { + return "", false + } + return strings.ToLower(m[1]), true +} + +// hasAmbiguousDecide reports whether a decision marker was attempted but could +// not be parsed, which the caller treats as "do not push" (fail safe). +func hasAmbiguousDecide(text string) bool { + if decideMarker.MatchString(text) { + return false + } + for _, p := range decidePrefixes { + if strings.Contains(text, p) { + return true + } + } + return false +} + +// matchChannel returns the lowercased target-channel token, or "" when the +// conclusion carries no (or a malformed) channel marker. A malformed channel +// marker degrades to the default route rather than failing the whole pass. +func matchChannel(text string) string { + m := channelMarker.FindStringSubmatch(text) + if m == nil { + return "" + } + return strings.ToLower(m[1]) +} + +// stripMarkers removes the structured decision/channel markers and any lines that +// become empty, leaving the clean message body. It tolerates the markers on their +// own lines or inline with the body. +func stripMarkers(text string) string { + lines := strings.Split(text, "\n") + out := make([]string, 0, len(lines)) + for _, ln := range lines { + ln = strings.TrimSpace(decideMarker.ReplaceAllString(ln, "")) + ln = strings.TrimSpace(channelMarker.ReplaceAllString(ln, "")) + if ln == "" { + continue + } + out = append(out, ln) + } + return strings.Join(out, "\n") +} + +// looksNone reports whether the body says "nothing worth saying". It is used both +// to drive the no-marker default and to fail-safe a contradictory push marker. +func looksNone(text string) bool { + t := normalizeNone(text) + if t == "" { + return false + } + for _, p := range nonePhrases { + if t == p { + return true + } + } + return false +} + +// normalizeNone lowercases the body and strips spaces and punctuation so the +// nonePhrases compare cleanly across "无", "无。", "Nothing to report." and similar. +func normalizeNone(text string) string { + var b strings.Builder + for _, r := range strings.ToLower(strings.TrimSpace(text)) { + if unicode.IsSpace(r) || unicode.IsPunct(r) { + continue + } + b.WriteRune(r) + } + return b.String() +} diff --git a/internal/services/backthink/proactive_test.go b/internal/services/backthink/proactive_test.go new file mode 100644 index 0000000..b7f84d6 --- /dev/null +++ b/internal/services/backthink/proactive_test.go @@ -0,0 +1,309 @@ +package backthink + +import ( + "context" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestParseProactive_WithMarkers(t *testing.T) { + t.Parallel() + conclusion := "[我决定:push]\n[渠道:kook]\n我想提醒你明天有体检" + msg, hint, push := parseProactive(conclusion) + if !push { + t.Fatalf("push = false, want true (decision said push)") + } + if hint != "kook" { + t.Fatalf("channel hint = %q, want kook", hint) + } + if msg != "我想提醒你明天有体检" { + t.Fatalf("message = %q, want the clean body", msg) + } +} + +func TestParseProactive_NoPushMarker(t *testing.T) { + t.Parallel() + conclusion := "[我决定:no_push]\n[渠道:telegram]\n暂时没有想说的" + msg, hint, push := parseProactive(conclusion) + if push { + t.Fatalf("push = true, want false (decision said no_push)") + } + if hint != "telegram" { + t.Fatalf("channel hint = %q, want telegram", hint) + } + if msg != "暂时没有想说的" { + t.Fatalf("message = %q, want the clean body", msg) + } +} + +func TestParseProactive_ChannelColonWidth(t *testing.T) { + t.Parallel() + // Tolerate a full-width colon and trailing spaces, and uppercase channel value. + conclusion := "[我决定：push] [渠道：WEB] 晚上早点睡" + msg, hint, push := parseProactive(conclusion) + if !push { + t.Fatalf("push = false, want true") + } + if hint != "web" { + t.Fatalf("channel hint = %q, want web (lowercased)", hint) + } + if msg != "晚上早点睡" { + t.Fatalf("message = %q, want the clean body", msg) + } +} + +func TestParseProactive_NothingVariants(t *testing.T) { + t.Parallel() + for _, c := range []string{ + "无", + "无。", + "nothing to report", + "Nothing to report.", + "nothing", + "暂无", + } { + if _, _, push := parseProactive(c); push { + t.Fatalf("parseProactive(%q) push = true, want false", c) + } + } +} + +func TestParseProactive_NoMarkerDefaultPush(t *testing.T) { + t.Parallel() + conclusion := "洗衣机洗好了，衣服可以收了" + msg, hint, push := parseProactive(conclusion) + if !push { + t.Fatalf("push = false, want true (no marker, not a nothing phrase)") + } + if hint != "" { + t.Fatalf("channel hint = %q, want empty", hint) + } + if msg != conclusion { + t.Fatalf("message = %q, want the whole conclusion", msg) + } +} + +func TestParseProactive_MalformedFailsSafe(t *testing.T) { + t.Parallel() + cases := []string{ + "[我决定:maybe]", // decision value outside allow-list + "[我决定]", // bracket without a value + "[我决定:push", // unterminated bracket + "[渠道:web]", // channel marker alone, no decision → but no decision marker + "[我决定:push", // deliberate ambiguity + } + for _, c := range cases { + if _, _, push := parseProactive(c); push { + t.Fatalf("parseProactive(%q) push = true, want false (fail safe on ambiguity)", c) + } + } +} + +func TestParseProactive_MarkerWithoutBody(t *testing.T) { + t.Parallel() + // A lone push marker with no body must not push an empty message (fail safe). + if _, _, push := parseProactive("[我决定:push]"); push { + t.Fatalf("push = true for a marker with no body, want false") + } + // A push marker whose body says "nothing" is contradictory → fail safe. + if _, _, push := parseProactive("[我决定:push]\n无"); push { + t.Fatalf("push = true for a push marker with nothing body, want false") + } +} + +func TestThink_ParsesStructuredDecision_EndToEnd(t *testing.T) { + t.Parallel() + conclusion := "[我决定:push]\n[渠道:kook]\n我想提醒你明天有体检" + thinker := &mockThinker{text: conclusion} + active := &mockActive{n: 1} + im := &mockIMAdapter{} + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: thinker, + State: &mockState{state: UserState{}}, + Active: active, + IM: im, + Gate: &mockGate{allow: true}, + Now: func() time.Time { return dayAt() }, + }) + + b.think(context.Background(), heartbeatTrigger()) + + // The channel hint "kook" routes to the IM bridge (a specific platform), not the + // active client. + if active.calls != 0 { + t.Fatalf("active calls = %d, want 0 (specific platform routes to IM)", active.calls) + } + if im.count() != 1 { + t.Fatalf("im calls = %d, want 1", im.count()) + } + if im.last.Content != "我想提醒你明天有体检" { + t.Fatalf("im message content = %q, want the clean body", im.last.Content) + } + if im.last.Platform != "kook" { + t.Fatalf("im message platform = %q, want kook", im.last.Platform) + } + + recs, err := b.List(context.Background(), 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) == 0 { + t.Fatal("no record persisted") + } + if !recs[0].ShouldPush { + t.Fatalf("ShouldPush = false, want true") + } + if recs[0].Conclusion != "我想提醒你明天有体检" { + t.Fatalf("record Conclusion = %q, want the clean body", recs[0].Conclusion) + } + if recs[0].PushResult.Route != RouteIMBridge { + t.Fatalf("record Route = %q, want %q", recs[0].PushResult.Route, RouteIMBridge) + } + if !strings.Contains(recs[0].PushResult.Detail, "channel_hint=kook") { + t.Fatalf("record Detail = %q, want it to include the kook hint", recs[0].PushResult.Detail) + } + if !recs[0].PushResult.Pushed { + t.Fatalf("record Pushed = false, want true") + } +} + +func TestThink_NoMarkerDefaultRoute_ActiveFirst(t *testing.T) { + t.Parallel() + // No channel marker → auto → active client first (the pre-P2 behavior). + conclusion := "[我决定:push]\n洗衣机洗好了，衣服可以收了" + thinker := &mockThinker{text: conclusion} + active := &mockActive{n: 2} + im := &mockIMAdapter{} + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: thinker, + State: &mockState{state: UserState{}}, + Active: active, + IM: im, + Gate: &mockGate{allow: true}, + Now: func() time.Time { return dayAt() }, + }) + + b.think(context.Background(), heartbeatTrigger()) + + if active.calls != 1 { + t.Fatalf("active calls = %d, want 1", active.calls) + } + if im.count() != 0 { + t.Fatalf("im calls = %d, want 0 (active preferred)", im.count()) + } + recs, err := b.List(context.Background(), 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if recs[0].PushResult.Route != RouteActiveClient { + t.Fatalf("record Route = %q, want %q", recs[0].PushResult.Route, RouteActiveClient) + } + if recs[0].PushResult.Pushed != true { + t.Fatalf("record Pushed = %v, want true", recs[0].PushResult.Pushed) + } +} + +func TestThink_ConcludedNoPush_NotPushed(t *testing.T) { + t.Parallel() + conclusion := "[我决定:no_push]\n今晚就不打扰她了" + thinker := &mockThinker{text: conclusion} + active := &mockActive{n: 1} + im := &mockIMAdapter{} + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: thinker, + State: &mockState{state: UserState{}}, + Active: active, + IM: im, + Gate: &mockGate{allow: true}, + Now: func() time.Time { return dayAt() }, + }) + + b.think(context.Background(), heartbeatTrigger()) + + if active.calls != 0 || im.count() != 0 { + t.Fatalf("push attempted despite no_push: active=%d im=%d", active.calls, im.count()) + } + recs, err := b.List(context.Background(), 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if recs[0].ShouldPush { + t.Fatalf("ShouldPush = true, want false") + } + if recs[0].PushResult.Route != RouteNone { + t.Fatalf("record Route = %q, want %q", recs[0].PushResult.Route, RouteNone) + } +} + +func TestThink_DoNotDisturbBlocksPush(t *testing.T) { + t.Parallel() + // She wants to push, but the user is asleep → the hard safety gate refuses, + // regardless of her decision. + conclusion := "[我决定:push]\n想跟你说晚安" + thinker := &mockThinker{text: conclusion} + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: thinker, + State: &mockState{state: UserState{Sleeping: true}}, + Pusher: &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}, + Now: func() time.Time { return nightAt() }, + }) + + b.think(context.Background(), heartbeatTrigger()) + + recs, err := b.List(context.Background(), 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if recs[0].ShouldPush { + t.Fatalf("ShouldPush = true, want false (asleep blocks push)") + } + if recs[0].PushResult.Route != RouteNone { + t.Fatalf("record Route = %q, want %q", recs[0].PushResult.Route, RouteNone) + } +} + +func TestBuildProactiveMessage_DefaultPlatform(t *testing.T) { + t.Parallel() + b := &Backthink{now: dayAt} + msg := b.buildProactiveMessage("晚安", "") + if msg.Platform != defaultProactivePlatform { + t.Fatalf("Platform = %q, want default %q", msg.Platform, defaultProactivePlatform) + } + if msg.Content != "晚安" { + t.Fatalf("Content = %q, want 晚安", msg.Content) + } + if msg.ChannelType != adapter.ChannelDirect || msg.ContentType != adapter.ContentText { + t.Fatalf("message kind = %q/%q, want direct/text", msg.ChannelType, msg.ContentType) + } +} + +func TestBuildProactiveMessage_SpecificPlatform(t *testing.T) { + t.Parallel() + b := &Backthink{now: dayAt} + msg := b.buildProactiveMessage("晚安", "telegram") + if msg.Platform != "telegram" { + t.Fatalf("Platform = %q, want telegram", msg.Platform) + } +} + +func TestPusher_SpecificPlatform_NoIM_FallsBackToActive(t *testing.T) { + t.Parallel() + // When a specific platform is requested but no IM sink is wired, the pusher + // falls back to the default route so the owner still receives the message, and + // records the requested hint. + active := &mockActive{n: 1} + p := newPusher(active, nil, &mockGate{allow: true}, discardLog()) + res := p.Push(context.Background(), PushRequest{ + Message: adapter.UnifiedMessage{}, + ChannelHint: "kook", + }) + if !res.Pushed || res.Route != RouteActiveClient { + t.Fatalf("push = %+v, want pushed active_client (fallback)", res) + } + if !strings.Contains(res.Detail, "channel_hint=kook") { + t.Fatalf("Detail = %q, want it to include the kook hint", res.Detail) + } +} diff --git a/internal/services/backthink/push.go b/internal/services/backthink/push.go index f8b9b74..a100f9c 100644 --- a/internal/services/backthink/push.go +++ b/internal/services/backthink/push.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "strings" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" @@ -25,6 +26,19 @@ const ( RouteIMBridge Route = "im_bridge" ) +// Channel-hint tokens recognized by the push router. They come from the +// conclusion's [渠道:...] marker (parseProactive) and select where a proactive +// message is delivered. An empty/unknown hint is treated as auto. A concrete +// platform id (kook/telegram/...) is a value other than these three. +const ( + channelAuto = "auto" + channelIM = "im" + channelWeb = "web" + // defaultProactivePlatform is the Platform the outbound proactive message + // carries when the conclusion names no specific channel. + defaultProactivePlatform = "cyrene-backthink" +) + // String returns the route as a plain label. func (r Route) String() string { return string(r) } @@ -75,16 +89,30 @@ type PushGate interface { Allow(ctx context.Context) bool } +// PushRequest carries one proactive push attempt: the outbound message plus the +// channel target the autonomous conclusion asked for (design/20 §2 P2). +// ChannelHint is the lowercased [渠道:...] token: ""/auto keep the default +// active-client-first, IM-fallback route; "web" targets the active client; "im" +// targets the IM bridge; a concrete platform id (kook/telegram/...) targets the IM +// bridge with the message tagged for that platform. +type PushRequest struct { + // Message is the normalized outbound proactive message. + Message adapter.UnifiedMessage + // ChannelHint is the requested target channel, "" when she did not choose one. + ChannelHint string +} + // Pusher delivers a proactive message under the active-client-first, IM-fallback -// policy (design/12 §3), bounded by the rate gate. +// policy (design/12 §3), bounded by the rate gate, honoring the channel hint the +// autonomous conclusion selected. type Pusher interface { - // Push delivers msg, preferring the active client and falling back to the IM - // bridge. It never blocks on a slow target. - Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult + // Push delivers req preferring the active client and falling back to the IM + // bridge, per the channel hint. It never blocks on a slow target. + Push(ctx context.Context, req PushRequest) PushResult } // pusher is the concrete Pusher: it consults the gate, then tries the active -// client and falls back to IM. +// client and falls back to IM, honoring the requested channel hint. type pusher struct { active ActiveClient im IMAdapter @@ -99,8 +127,11 @@ func newPusher(active ActiveClient, im IMAdapter, gate PushGate, log *slog.Logge return &pusher{active: active, im: im, gate: gate, log: log} } -// Push implements Pusher. -func (p *pusher) Push(ctx context.Context, msg adapter.UnifiedMessage) PushResult { +// Push implements Pusher: it applies the rate gate, then resolves the channel +// hint to a route. A specific platform targets the IM bridge (tagged on the +// message) and falls back to the default route when no IM sink is wired, so the +// owner still receives the message rather than having it dropped. +func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult { if err := ctx.Err(); err != nil { return PushResult{Pushed: false, Route: RouteNone, Err: err.Error()} } @@ -109,7 +140,36 @@ func (p *pusher) Push(ctx context.Context, msg adapter.UnifiedMessage) PushResul return PushResult{Pushed: false, Route: RouteRateLimited, Detail: "proactive push rate limited"} } - // Active client first. + hint := normalizeChannelHint(req.ChannelHint) + msg := req.Message + // A specific platform names the requested channel on the message so a + // platform-aware IM sink can honor it; auto/web/im keep the default platform. + if isSpecificChannel(hint) { + msg.Platform = hint + } + + var res PushResult + switch hint { + case channelWeb: + res = p.pushActive(ctx, msg) + case channelIM: + res = p.pushIM(ctx, msg) + case channelAuto: + res = p.pushDefault(ctx, msg) + default: + // A specific platform targets the IM bridge for that channel. With no IM + // sink wired (the backthink IM seam is single-sink), fall back to the + // default route so the owner is still reached; the hint is recorded. + res = p.pushIM(ctx, msg) + if !res.Pushed { + res = p.pushDefault(ctx, msg) + } + } + return annotateHint(res, req.ChannelHint) +} + +// pushDefault tries the active client first and falls back to the IM bridge. +func (p *pusher) pushDefault(ctx context.Context, msg adapter.UnifiedMessage) PushResult { if p.active != nil { n, err := p.active.SendProactive(ctx, msg) if err != nil { @@ -118,17 +178,74 @@ func (p *pusher) Push(ctx context.Context, msg adapter.UnifiedMessage) PushResul return PushResult{Pushed: true, Route: RouteActiveClient, Detail: fmt.Sprintf("delivered to %d active client(s)", n)} } } - - // IM fallback. - if p.im != nil { - if err := p.im.SendProactive(ctx, msg); err != nil { - p.log.Warn("backthink IM push failed", "error", err) - return PushResult{Pushed: false, Route: RouteIMBridge, Err: err.Error()} - } - return PushResult{Pushed: true, Route: RouteIMBridge, Detail: "delivered via IM bridge"} + if p.im == nil { + return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client or IM target"} } + return p.pushIM(ctx, msg) +} - return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client or IM target"} +// pushActive delivers only to the active web/native client. +func (p *pusher) pushActive(ctx context.Context, msg adapter.UnifiedMessage) PushResult { + if p.active != nil { + n, err := p.active.SendProactive(ctx, msg) + if err != nil { + p.log.Warn("backthink active client push failed", "error", err) + } else if n > 0 { + return PushResult{Pushed: true, Route: RouteActiveClient, Detail: fmt.Sprintf("delivered to %d active client(s)", n)} + } + } + return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client"} +} + +// pushIM delivers only to the IM bridge. +func (p *pusher) pushIM(ctx context.Context, msg adapter.UnifiedMessage) PushResult { + if p.im == nil { + return PushResult{Pushed: false, Route: RouteNone, Detail: "no IM target"} + } + if err := p.im.SendProactive(ctx, msg); err != nil { + p.log.Warn("backthink IM push failed", "error", err) + return PushResult{Pushed: false, Route: RouteIMBridge, Err: err.Error()} + } + return PushResult{Pushed: true, Route: RouteIMBridge, Detail: "delivered via IM bridge"} +} + +// normalizeChannelHint lowercases and trims a channel hint, mapping synonyms to +// the canonical route tokens (auto/im/web); anything else is a concrete platform +// id (kook/telegram/...). Unknown/blank maps to auto. +func normalizeChannelHint(hint string) string { + switch strings.ToLower(strings.TrimSpace(hint)) { + case "", "auto": + return channelAuto + case "im", "bridge": + return channelIM + case "web", "frontend", "active": + return channelWeb + default: + return strings.ToLower(strings.TrimSpace(hint)) + } +} + +// isSpecificChannel reports whether hint names a concrete platform rather than a +// routing keyword (auto/im/web). +func isSpecificChannel(hint string) bool { + return hint != channelAuto && hint != channelIM && hint != channelWeb +} + +// annotateHint appends the requested channel hint to the push result detail so +// the record/panel surfaces which channel she chose, even when the route fell +// back. Blank/auto hints add nothing. +func annotateHint(res PushResult, hint string) PushResult { + h := strings.ToLower(strings.TrimSpace(hint)) + if h == "" || h == channelAuto { + return res + } + tag := "channel_hint=" + h + if res.Detail == "" { + res.Detail = tag + } else { + res.Detail = res.Detail + ", " + tag + } + return res } // rateGate is the production PushGate: a sliding-window counter over @@ -153,13 +270,19 @@ func (g *rateGate) Allow(ctx context.Context) bool { } // buildProactiveMessage assembles the normalized outbound message carrying a -// think conclusion. It is intentionally minimal: the concrete active-client and -// IM adapters apply their own routing/targeting to the owner. -func (b *Backthink) buildProactiveMessage(conclusion string) adapter.UnifiedMessage { +// think conclusion's clean body. It tags the platform with the requested channel +// when the conclusion names a concrete platform, otherwise the default backthink +// platform. The concrete active-client and IM adapters apply their own +// routing/targeting to the owner (red line: owner only). +func (b *Backthink) buildProactiveMessage(message, channelHint string) adapter.UnifiedMessage { + platform := defaultProactivePlatform + if h := normalizeChannelHint(channelHint); isSpecificChannel(h) { + platform = h + } return adapter.UnifiedMessage{ - Platform: "cyrene-backthink", + Platform: platform, ChannelType: adapter.ChannelDirect, - Content: conclusion, + Content: message, ContentType: adapter.ContentText, Sender: adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"}, Timestamp: b.now(), diff --git a/internal/services/backthink/push_test.go b/internal/services/backthink/push_test.go index d2bf7ff..6bf1b81 100644 --- a/internal/services/backthink/push_test.go +++ b/internal/services/backthink/push_test.go @@ -14,7 +14,7 @@ func TestPusher_ActiveFirst(t *testing.T) { im := &mockIMAdapter{} p := newPusher(active, im, &mockGate{allow: true}, discardLog()) - res := p.Push(context.Background(), adapter.UnifiedMessage{}) + res := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) if !res.Pushed { t.Fatalf("Pushed = false, want true") @@ -36,7 +36,7 @@ func TestPusher_IMFallback_NoActive(t *testing.T) { im := &mockIMAdapter{} p := newPusher(active, im, &mockGate{allow: true}, discardLog()) - res := p.Push(context.Background(), adapter.UnifiedMessage{}) + res := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) if !res.Pushed { t.Fatalf("Pushed = false, want true") @@ -55,7 +55,7 @@ func TestPusher_IMFallback_ActiveError(t *testing.T) { im := &mockIMAdapter{} p := newPusher(active, im, &mockGate{allow: true}, discardLog()) - res := p.Push(context.Background(), adapter.UnifiedMessage{}) + res := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) if !res.Pushed { t.Fatalf("Pushed = false, want true") @@ -72,7 +72,7 @@ func TestPusher_NoTarget_Skipped(t *testing.T) { t.Parallel() p := newPusher(nil, nil, &mockGate{allow: true}, discardLog()) - res := p.Push(context.Background(), adapter.UnifiedMessage{}) + res := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) if res.Pushed { t.Fatalf("Pushed = true, want false") @@ -88,7 +88,7 @@ func TestPusher_GateDenied(t *testing.T) { im := &mockIMAdapter{} p := newPusher(active, im, &mockGate{allow: false}, discardLog()) - res := p.Push(context.Background(), adapter.UnifiedMessage{}) + res := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) if res.Pushed { t.Fatalf("Pushed = true, want false") @@ -108,7 +108,7 @@ func TestPusher_CancelledContext_Denied(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - res := p.Push(ctx, adapter.UnifiedMessage{}) + res := p.Push(ctx, PushRequest{Message: adapter.UnifiedMessage{}}) if res.Pushed { t.Fatalf("Pushed = true, want false") @@ -128,9 +128,9 @@ func TestPusher_RateLimit(t *testing.T) { active := &mockActive{n: 1} p := newPusher(active, nil, newRateGate(1, time.Minute, discardLog()), discardLog()) - first := p.Push(context.Background(), adapter.UnifiedMessage{}) - second := p.Push(context.Background(), adapter.UnifiedMessage{}) - third := p.Push(context.Background(), adapter.UnifiedMessage{}) + first := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) + second := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) + third := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{}}) if !first.Pushed || first.Route != RouteActiveClient { t.Fatalf("first push = %+v, want active_client pushed", first)