feat(backthink): 自主思考 P2——结构化推送决策+渠道选择(design/20)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 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接口限制
This commit is contained in:
Aska committed 2026-09-07 19:34:51 +08:00
1 parent 93aff2fea5
commit 2283383b9e
8 files changed
+688 -47

No files matched your search

+25 -10
View File
@@ -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)
}
}
+10
View File
@@ -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" +
"整轮思考保持简短、单轮,不反复自我触发。"
}
+28
View File
@@ -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.
+8 -6
View File
@@ -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
}
+154
View File
@@ -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 warm message body>
//
// 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()
}
@@ -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)
}
}
+145 -22
View File
@@ -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(),
+9 -9
View File
@@ -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)