From 9b207a31e82bc89ccd06af692f6fc05a9273a0b0 Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 15:53:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20=E5=9C=B0=E5=9F=BA=E2=91=A2=20?= =?UTF-8?q?=E6=89=93=E9=80=9A=E5=AE=9E=E6=97=B6=E4=BF=A1=E5=8F=B7=20produc?= =?UTF-8?q?er=E2=86=92=E5=89=8D=E7=AB=AFWS=20+=20loop=20=E7=9C=9F=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=20+=20=E4=B8=8A=E8=A1=8C=E8=B7=AF=E7=94=B1=E8=A1=A5?= =?UTF-8?q?=E5=85=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 打通前端 ws.ts 里那批死消费者事件（thinking/mood/tool_used/subagent_event/ task_output/device_update/reminder/backthink），并给 loop 接真 token 流式。 - wire.go：loop 建时传 OnStream（streamChunk 闭包把 delta 发 bus ws.stream_chunk）； gateway.Deps 注入 Scheduler/Tasks/Streaming；iot 接共享总线 iot.WithBus(a.bus)。 - ws_bridge.go：hub 额外订阅 iot.>/task.>/automation.>/reminder.>/cyrene.agent.*.event， normalizeEvent 归一为 ws.*（device_update/task_output/task_done/automation_result/ reminder/subagent_event）；不订阅 cyrene.agent.> 以免 ask/reply 内部事件进 relay。 - ws_route.go：voice_input→ws.voice_input；subagent_ask→Scheduler.AuditContext（真实 上下文审计，替换 pending stub）；task_output_query→Manager.TaskOutput（真实输出， 替换空 {}）。 - server.go：新增 SubagentAuditor/TaskOutputStore 接口 + Deps.Scheduler/Tasks/Streaming。 - orchestration/auditcontext.go：Scheduler.AuditContext(full/tail/keyword)。 - task/manager.go：Manager.TaskOutput(id)。 - backthink：record() 补发 ws.backthink 事件。 - json tag 修正（device_update/reminder/task output 结构体加小写 tag + DeviceUpdate 补 message），使前端分支能真正渲染。 实机验证：WS 握手 101、task manager ready、iot manager initialized(WithBus 生效)。 go build/vet/test 全绿；TestHubBridgeDeviceUpdate/TestHubBridgeSubagentProgress 覆盖 映射。dev 推至后续版本。 --- cmd/cyrene/stream.go | 22 ++++ cmd/cyrene/wire.go | 28 ++-- internal/iot/device.go | 11 +- internal/iot/manager.go | 18 ++- .../runtime/orchestration/auditcontext.go | 94 ++++++++++++++ .../orchestration/auditcontext_test.go | 57 +++++++++ internal/runtime/task/manager.go | 49 +++++-- internal/services/automations/reminder.go | 6 +- internal/services/backthink/backthink.go | 7 + internal/services/gateway/extra_test.go | 120 ++++++++++++++++-- internal/services/gateway/server.go | 76 +++++++---- internal/services/gateway/ws_bridge.go | 92 ++++++++++++++ internal/services/gateway/ws_hub.go | 95 ++++++++------ internal/services/gateway/ws_route.go | 107 ++++++++++++++++ 14 files changed, 673 insertions(+), 109 deletions(-) create mode 100644 cmd/cyrene/stream.go create mode 100644 internal/runtime/orchestration/auditcontext.go create mode 100644 internal/runtime/orchestration/auditcontext_test.go create mode 100644 internal/services/gateway/ws_bridge.go create mode 100644 internal/services/gateway/ws_route.go diff --git a/cmd/cyrene/stream.go b/cmd/cyrene/stream.go new file mode 100644 index 0000000..f35d768 --- /dev/null +++ b/cmd/cyrene/stream.go @@ -0,0 +1,22 @@ +package main + +import "context" + +// streamChunkSubject is the bus subject the loop's OnStream closure publishes +// each token delta to. The gateway hub subscribes to ws.> and re-broadcasts it +// as a ws.stream_chunk frame to every client. +const streamChunkSubject = "ws.stream_chunk" + +// streamChunk is the loop's OnStream callback. It publishes one token delta to +// the gateway's streaming subject so the hub fans it out to every client as a +// ws.stream_chunk frame. It is a no-op when no bus is wired and never blocks the +// loop: bus.Publish is asynchronous and a full per-subscription buffer drops the +// event rather than stalling the model stream. +func (a *app) streamChunk(delta string) { + if a.bus == nil { + return + } + if err := a.bus.Publish(context.Background(), streamChunkSubject, map[string]any{"delta": delta}); err != nil { + a.log.Warn("stream chunk publish", "err", err) + } +} diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 847edca..7fa5591 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -44,8 +44,8 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/telegram" "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/webhook" "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink" - iotools "git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/io" "git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/basic" + iotools "git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/io" ) // defaultOwnerPath and defaultPersonasDir point at the on-disk persona/owner @@ -302,6 +302,7 @@ func (a *app) wireRuntime() error { PersonaID: personaID, MaxRounds: defaultMaxRounds, WorkingMemory: memory.NewWorkingMemory(0), + OnStream: a.streamChunk, Log: logger.WithSource(a.log, "runtime/loop"), }) @@ -331,7 +332,7 @@ func (a *app) wireAuxiliary() error { } a.tasks = tmgr - iotMgr, err := iot.New(a.cfg.IoT, a.log) + iotMgr, err := iot.New(a.cfg.IoT, a.log, iot.WithBus(a.bus)) if err != nil { return fmt.Errorf("iot: %w", err) } @@ -404,16 +405,19 @@ func (a *app) wireChannels() error { func (a *app) wireFrontend() error { jwt := auth.New(a.cfg.Auth) gw, err := gateway.New(a.cfg, gateway.Deps{ - Log: logger.WithSource(a.log, "gateway"), - Auth: jwt, - Bus: a.bus, - Registry: a.caps, - Loop: a.loop, - Session: a.sessions, - Personas: a.personas, - Memories: a.memStore, - IoT: a.iot, - Frontend: resolveFrontendFS(a.log), + Log: logger.WithSource(a.log, "gateway"), + Auth: jwt, + Bus: a.bus, + Registry: a.caps, + Loop: a.loop, + Session: a.sessions, + Personas: a.personas, + Memories: a.memStore, + IoT: a.iot, + Scheduler: a.scheduler, + Tasks: a.tasks, + Streaming: true, + Frontend: resolveFrontendFS(a.log), }) if err != nil { return fmt.Errorf("gateway: %w", err) diff --git a/internal/iot/device.go b/internal/iot/device.go index 23d37ed..6e4de22 100644 --- a/internal/iot/device.go +++ b/internal/iot/device.go @@ -82,11 +82,14 @@ func (d Device) Clone() Device { // state transitions, typically as the result of a Control call. type DeviceUpdate struct { // Provider is the name of the provider that owns the device. - Provider string + Provider string `json:"provider"` // Device is the post-transition device snapshot. - Device Device + Device Device `json:"device"` // Action is the control action that produced the transition. - Action string + Action string `json:"action"` // Params are the optional control parameters. - Params map[string]any + Params map[string]any `json:"params,omitempty"` + // Message is a human-readable summary of the transition, surfaced to the + // frontend device_update feed. + Message string `json:"message,omitempty"` } diff --git a/internal/iot/manager.go b/internal/iot/manager.go index d634391..0682bb3 100644 --- a/internal/iot/manager.go +++ b/internal/iot/manager.go @@ -22,6 +22,17 @@ type Manager struct { log *slog.Logger } +// Option customizes a Manager built by New. +type Option func(*Manager) + +// WithBus injects an existing bus instead of letting the Manager create one. It +// mirrors the other subsystems so iot.device_update flows on the shared bus and +// reaches the gateway hub / background-trigger subscribers rather than staying +// on an isolated manager-owned bus. +func WithBus(b bus.Bus) Option { + return func(m *Manager) { m.bus = b } +} + // New builds a Manager from cfg, selecting the device Provider by cfg.Provider. // // Supported providers: "" or "mock" registers the built-in mock provider with @@ -31,11 +42,14 @@ type Manager struct { // // log may be nil; it is replaced with a discard logger. The Manager owns a // fresh, isolated bus exposed via Bus for subscribers to attach to. -func New(cfg config.IoTConfig, log *slog.Logger) (*Manager, error) { +func New(cfg config.IoTConfig, log *slog.Logger, opts ...Option) (*Manager, error) { m := &Manager{ bus: bus.New(), log: logger.WithSource(discardLogger(log), "iot/manager"), } + for _, opt := range opts { + opt(m) + } switch strings.TrimSpace(cfg.Provider) { case "", "mock": mp, err := newMockProvider(log) @@ -142,11 +156,13 @@ func (m *Manager) Status(ctx context.Context, deviceID string) (*Device, error) // changed, and control should not be reported as failed because an observer did // not receive the event. func (m *Manager) publishUpdate(ctx context.Context, provider string, d Device, action string, params map[string]any) { + msg := fmt.Sprintf("设备 %s · %s（%s）", d.Name, action, provider) if err := m.bus.Publish(ctx, DeviceUpdateSubject, DeviceUpdate{ Provider: provider, Device: d, Action: action, Params: params, + Message: msg, }); err != nil { m.log.Error("iot publish device update", "device", d.ID, "action", action, "error", err) return diff --git a/internal/runtime/orchestration/auditcontext.go b/internal/runtime/orchestration/auditcontext.go new file mode 100644 index 0000000..e11761d --- /dev/null +++ b/internal/runtime/orchestration/auditcontext.go @@ -0,0 +1,94 @@ +package orchestration + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/audit" +) + +// AuditMode selects how a sub-agent context is shaped for the subagent_ask +// upstream WS route. It mirrors the audit.ContextAuditor surface (Full/Tail/ +// Keyword) so the gateway can request exactly the window the client asks for. +type AuditMode string + +const ( + // AuditFull returns the entire conversation context. + AuditFull AuditMode = "full" + // AuditTail returns the last n messages of the context. + AuditTail AuditMode = "tail" + // AuditKeyword returns the window around the first keyword match. + AuditKeyword AuditMode = "keyword" +) + +// AuditResult is the read-only result of auditing a sub-agent's conversation +// context. Messages carries the selected window; Line and Match are populated for +// a keyword query. It is the wire payload for a ws.subagent_event reply. +type AuditResult struct { + AgentID string `json:"agentId"` + Status Status `json:"status"` + Goal string `json:"goal,omitempty"` + Mode AuditMode `json:"mode"` + Messages []model.Message `json:"messages,omitempty"` + Line int `json:"line,omitempty"` + Match string `json:"match,omitempty"` +} + +// AuditContext returns a read-only view of sub-agent id's conversation context +// for the subagent_ask WS route. mode selects the window: AuditFull returns the +// whole context, AuditTail returns the last n messages, AuditKeyword returns the +// window around the first query hit. It wraps ErrAgentNotFound for an unknown id +// and audit.ErrQueryEmpty for a blank keyword query, both with %w so callers can +// classify with errors.Is. +func (s *Scheduler) AuditContext(ctx context.Context, id string, mode AuditMode, n int, query string) (*AuditResult, error) { + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("orchestration: audit %q: %w", id, err) + } + s.mu.Lock() + sub := s.agents[id] + s.mu.Unlock() + if sub == nil { + return nil, fmt.Errorf("orchestration: audit %q: %w", id, ErrAgentNotFound) + } + + a := audit.NewContextAuditor(sub.Context) + res := &AuditResult{AgentID: sub.ID, Status: sub.Status(), Goal: sub.Goal, Mode: mode} + switch mode { + case AuditFull: + res.Messages = a.Full() + case AuditTail: + res.Messages = a.Tail(n) + case AuditKeyword: + loc, err := a.Keyword(ctx, query) + if err != nil { + return nil, fmt.Errorf("orchestration: audit %q: %w", id, err) + } + if loc == nil { + res.Messages = []model.Message{} + break + } + res.Line = loc.Line + res.Messages = loc.Surrounding + res.Match = messageText(loc.Surrounding[len(loc.Surrounding)-1]) + default: + return nil, fmt.Errorf("orchestration: audit %q: unknown mode %q", id, mode) + } + return res, nil +} + +// messageText concatenates the text parts of one message. It mirrors the audit +// subpackage's own helper so the result payload can surface a human-readable +// match without exporting a cross-package text extractor. +func messageText(m model.Message) string { + var text string + for _, c := range m.Content { + if c.Type == model.ContentTypeText && c.Text != "" { + if text != "" { + text += " " + } + text += c.Text + } + } + return text +} diff --git a/internal/runtime/orchestration/auditcontext_test.go b/internal/runtime/orchestration/auditcontext_test.go new file mode 100644 index 0000000..d15ac60 --- /dev/null +++ b/internal/runtime/orchestration/auditcontext_test.go @@ -0,0 +1,57 @@ +package orchestration + +import ( + "context" + "errors" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/audit" +) + +// TestSchedulerAuditContext exercises the subagent_ask audit path across the full, +// tail and keyword windows, plus the not-found and blank-query errors. +func TestSchedulerAuditContext(t *testing.T) { + t.Parallel() + s := New(Options{MaxParallel: 2, AgentID: "orchestrator"}) + ctx := context.Background() + + msgs := []model.Message{ + model.TextMessage(model.RoleUser, "hello"), + model.TextMessage(model.RoleAssistant, "hi there"), + } + s.mu.Lock() + s.agents["sub-1"] = &SubAgent{ID: "sub-1", Goal: "do it", Context: msgs, status: StatusRunning} + s.mu.Unlock() + + full, err := s.AuditContext(ctx, "sub-1", AuditFull, 0, "") + if err != nil { + t.Fatalf("audit full: %v", err) + } + if full.AgentID != "sub-1" || full.Status != StatusRunning || len(full.Messages) != 2 { + t.Fatalf("full = %+v, want sub-1/running/2 msgs", full) + } + + tail, err := s.AuditContext(ctx, "sub-1", AuditTail, 1, "") + if err != nil { + t.Fatalf("audit tail: %v", err) + } + if len(tail.Messages) != 1 || tail.Messages[0].Content[0].Text != "hi there" { + t.Fatalf("tail = %+v, want last message 'hi there'", tail) + } + + kw, err := s.AuditContext(ctx, "sub-1", AuditKeyword, 0, "hello") + if err != nil { + t.Fatalf("audit keyword: %v", err) + } + if kw.Line != 1 || len(kw.Messages) != 2 { + t.Fatalf("keyword = %+v, want line 1 with 2 surrounding msgs", kw) + } + + if _, err := s.AuditContext(ctx, "nope", AuditFull, 0, ""); !errors.Is(err, ErrAgentNotFound) { + t.Fatalf("want ErrAgentNotFound, got %v", err) + } + if _, err := s.AuditContext(ctx, "sub-1", AuditKeyword, 0, " "); !errors.Is(err, audit.ErrQueryEmpty) { + t.Fatalf("want ErrQueryEmpty, got %v", err) + } +} diff --git a/internal/runtime/task/manager.go b/internal/runtime/task/manager.go index 81398b4..e54ccd3 100644 --- a/internal/runtime/task/manager.go +++ b/internal/runtime/task/manager.go @@ -43,25 +43,25 @@ var ( // reaches a terminal state. It carries everything the agent needs to consume the // result without a second lookup. type DoneEvent struct { - ID string - Kind Kind - Status Status - Output Output - ExitCode int - Truncated bool - Error string - StartedAt time.Time - FinishedAt time.Time + ID string `json:"id"` + Kind Kind `json:"kind"` + Status Status `json:"status"` + Output Output `json:"output"` + ExitCode int `json:"exitCode"` + Truncated bool `json:"truncated"` + Error string `json:"error,omitempty"` + StartedAt time.Time `json:"startedAt"` + FinishedAt time.Time `json:"finishedAt"` } // OutputEvent is the payload published to the "task.output" subject as output // becomes available. type OutputEvent struct { - ID string - Kind Kind - Status Status - Output Output - Truncated bool + ID string `json:"id"` + Kind Kind `json:"kind"` + Status Status `json:"status"` + Output Output `json:"output"` + Truncated bool `json:"truncated"` } // Manager owns the task registry and the execution lifecycle. It owns an @@ -230,6 +230,27 @@ func (m *Manager) Get(ctx context.Context, id string) (*Task, error) { return t, nil } +// TaskOutput returns the current output snapshot of task id for external +// consumers (the gateway's task_output_query WS route). It never waits for the +// task to finish: for a running background task it returns the output captured so +// far. It wraps ErrNotFound with %w so callers can classify with errors.Is. +func (m *Manager) TaskOutput(ctx context.Context, id string) (OutputEvent, error) { + if err := ctx.Err(); err != nil { + return OutputEvent{}, fmt.Errorf("task: output %q: %w", id, err) + } + t := m.get(id) + if t == nil { + return OutputEvent{}, fmt.Errorf("task: output %q: %w", id, ErrNotFound) + } + return OutputEvent{ + ID: t.ID, + Kind: t.Kind, + Status: t.Status(), + Output: t.Output(), + Truncated: t.Truncated(), + }, nil +} + // Abort cancels the task for id. It returns ErrNotFound if there is no such task // and ErrAlreadyDone if the task has already finished. func (m *Manager) Abort(ctx context.Context, id string) error { diff --git a/internal/services/automations/reminder.go b/internal/services/automations/reminder.go index 01b49c0..500ea91 100644 --- a/internal/services/automations/reminder.go +++ b/internal/services/automations/reminder.go @@ -22,11 +22,11 @@ const ( // Reminder is a single delayed notification. type Reminder struct { // ID uniquely identifies the reminder. - ID string + ID string `json:"id"` // At is the time the reminder should fire. - At time.Time + At time.Time `json:"at"` // Message is the payload surfaced to the user when it fires. - Message string + Message string `json:"message"` } // ReminderStore schedules delayed notifications and publishes each due reminder diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index e39bf28..9ac9fb9 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -397,6 +397,13 @@ func (b *Backthink) record(r Record) { if _, err := b.store.Append(context.Background(), r); err != nil { b.log.Error("backthink record failed", "error", err) } + // Surface the persisted pass on the WS stream so the frontend's backthink + // consumer (or its default fallback) observes it without polling the store. + if b.deps.Bus != nil { + if err := b.deps.Bus.Publish(context.Background(), "ws.backthink", r); err != nil { + b.log.Warn("backthink bus publish failed", "error", err) + } + } } // normalize fills a Config's zero fields with safe defaults. diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 8bef408..b1d2b34 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -25,6 +25,9 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task" "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" ) @@ -192,11 +195,31 @@ type stubIoT struct { func (s *stubIoT) List(ctx context.Context) ([]iot.Device, error) { return s.list, s.err } +// stubAuditor mimics SubagentAuditor for the subagent_ask route. +type stubAuditor struct { + res *orchestration.AuditResult + err error +} + +func (s *stubAuditor) AuditContext(ctx context.Context, id string, mode orchestration.AuditMode, n int, query string) (*orchestration.AuditResult, error) { + return s.res, s.err +} + +// stubTaskOutputs mimics TaskOutputStore for the task_output_query route. +type stubTaskOutputs struct { + ev task.OutputEvent + err error +} + +func (s *stubTaskOutputs) TaskOutput(ctx context.Context, id string) (task.OutputEvent, error) { + return s.ev, s.err +} + // stubChannel implements sdk.Channel. type stubChannel struct{} func (stubChannel) Name() string { return "qq" } -func (stubChannel) Connect(ctx context.Context) error { return nil } +func (stubChannel) Connect(ctx context.Context) error { return nil } func (stubChannel) Disconnect(ctx context.Context) error { return nil } func (stubChannel) IsConnected() bool { return true } func (stubChannel) HealthCheck(ctx context.Context) error { return nil } @@ -205,9 +228,9 @@ var _ sdk.Channel = stubChannel{} // captureRunner records the sessionID each Run call sees. type captureRunner struct { - got []string - res *loop.Result - err error + got []string + res *loop.Result + err error } func (c *captureRunner) Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) { @@ -664,8 +687,8 @@ func TestStatusRecorderFlush(t *testing.T) { func TestSPA(t *testing.T) { t.Parallel() front := fstest.MapFS{ - "index.html": &fstest.MapFile{Data: []byte("root")}, - "assets/app.js": &fstest.MapFile{Data: []byte("var x = 1;")}, + "index.html": &fstest.MapFile{Data: []byte("root")}, + "assets/app.js": &fstest.MapFile{Data: []byte("var x = 1;")}, } srv, ts, _ := buildEnv(t, cfgDefault(), Deps{Frontend: front}) _ = srv @@ -825,20 +848,48 @@ func TestHubRelayChatBranch(t *testing.T) { func TestHubRoute(t *testing.T) { t.Parallel() hb := newTestHub(nil, bus.New()) + hb.auditor = &stubAuditor{res: &orchestration.AuditResult{ + AgentID: "sub-1", Status: orchestration.StatusRunning, Goal: "do it", Mode: orchestration.AuditFull, + Messages: []model.Message{model.TextMessage(model.RoleUser, "hi")}, + }} + hb.taskOutputs = &stubTaskOutputs{ev: task.OutputEvent{ + ID: "task-1", Status: task.StatusDone, Output: task.Output{Stdout: "out", Stderr: "err"}, + }} c := newFakeConn(hb, 32) hb.register(c) - // subagent_ask → subagent_event (enqueueJSON). - c.route(wsInbound{Type: "subagent_ask"}) + // subagent_ask → subagent_event (real audit data, not a hardcoded pending). + c.route(wsInbound{Type: "subagent_ask", Data: json.RawMessage(`{"id":"sub-1"}`)}) if ev := readFakeFrame(t, c); ev.Type != "subagent_event" { t.Fatalf("subagent frame type = %q, want subagent_event", ev.Type) } - // task_output_query → task_output (enqueueJSON). - c.route(wsInbound{Type: "task_output_query"}) + // subagent_ask with no auditor → clear error, never a fabricated pending. + hb.auditor = nil + c.route(wsInbound{Type: "subagent_ask", Data: json.RawMessage(`{"id":"sub-1"}`)}) + if ev := readFakeFrame(t, c); ev.Type != "error" { + t.Fatalf("no-auditor subagent frame type = %q, want error", ev.Type) + } + hb.auditor = &stubAuditor{res: &orchestration.AuditResult{}} + + // task_output_query → task_output with real output. + c.route(wsInbound{Type: "task_output_query", Data: json.RawMessage(`{"taskId":"task-1"}`)}) if ev := readFakeFrame(t, c); ev.Type != "task_output" { t.Fatalf("task frame type = %q, want task_output", ev.Type) } - // unknown → error (enqueueJSON), with non-nil rawData via mode_switch path already. + // task_output_query with no task store → clear error, never an empty {}. + hb.taskOutputs = nil + c.route(wsInbound{Type: "task_output_query", Data: json.RawMessage(`{"taskId":"task-1"}`)}) + if ev := readFakeFrame(t, c); ev.Type != "error" { + t.Fatalf("no-task store frame type = %q, want error", ev.Type) + } + hb.taskOutputs = &stubTaskOutputs{ev: task.OutputEvent{ID: "task-1", Status: task.StatusDone, Output: task.Output{Stdout: "out"}}} + + // voice_input → publishes to bus (no frame enqueued, not an unknown error). + c.route(wsInbound{Type: "voice_input", Data: json.RawMessage(`{"audio":"x"}`)}) + if err := connHasNoFrame(c); err != nil { + t.Fatalf("voice_input should not enqueue a frame directly: %v", err) + } + // unknown → error (enqueueJSON). c.route(wsInbound{Type: "nope"}) if ev := readFakeFrame(t, c); ev.Type != "error" { t.Fatalf("unknown frame type = %q, want error", ev.Type) @@ -850,6 +901,53 @@ func TestHubRoute(t *testing.T) { } } +func TestHubBridgeDeviceUpdate(t *testing.T) { + t.Parallel() + b := bus.New() + hb := newTestHub(nil, b) + c := newFakeConn(hb, 32) + hb.register(c) + hb.start() + t.Cleanup(hb.close) + + if err := b.Publish(context.Background(), "iot.device_update", map[string]any{"message": "灯开了"}); err != nil { + t.Fatalf("publish: %v", err) + } + if ev := readFakeFrame(t, c); ev.Type != "device_update" { + t.Fatalf("type = %q, want device_update (frame=%+v)", ev.Type, ev) + } +} + +func TestHubBridgeSubagentProgress(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + c := newFakeConn(hb, 32) + hb.register(c) + + if err := hb.relay(context.Background(), bus.Event{ + Subject: "cyrene.agent.orchestrator.event", + Payload: comm.Event{Name: orchestration.EventDone, Data: orchestration.Progress{ + AgentID: "sub-1", Status: orchestration.StatusDone, RoleID: "engineer", Mode: "chat", + }}, + }); err != nil { + t.Fatalf("relay: %v", err) + } + if ev := readFakeFrame(t, c); ev.Type != "subagent_event" { + t.Fatalf("type = %q, want subagent_event (frame=%+v)", ev.Type, ev) + } + + // An internal ask subject must never surface to clients. + if err := hb.relay(context.Background(), bus.Event{ + Subject: "cyrene.agent.a.b.ask", + Payload: comm.Event{Name: orchestration.EventDone, Data: orchestration.Progress{AgentID: "sub-1"}}, + }); err != nil { + t.Fatalf("relay ask: %v", err) + } + if err := connHasNoFrame(c); err != nil { + t.Fatalf("internal ask must not be broadcast: %v", err) + } +} + func TestPublishPublicNilBus(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) // nil bus diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index bca5307..f40e2ff 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -38,6 +38,8 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task" ) // source is the log provenance tag for every record this package emits. @@ -96,6 +98,18 @@ type IoTStore interface { List(ctx context.Context) ([]iot.Device, error) } +// SubagentAuditor audits a sub-agent's conversation context for the subagent_ask +// WS route. *orchestration.Scheduler is the production implementation. +type SubagentAuditor interface { + AuditContext(ctx context.Context, id string, mode orchestration.AuditMode, n int, query string) (*orchestration.AuditResult, error) +} + +// TaskOutputStore queries a task's output for the task_output_query WS route. +// *task.Manager is the production implementation. +type TaskOutputStore interface { + TaskOutput(ctx context.Context, id string) (task.OutputEvent, error) +} + // Deps carries the gateway's injected dependencies. Every field is nil-tolerant // except Auth, which is required (authentication backs login and every RBAC // decision). Registry/Loop are satisfied by the concrete *capability.Registry and @@ -121,6 +135,16 @@ type Deps struct { Memories MemoryStore // IoT lists IoT devices. Optional. IoT IoTStore + // Scheduler audits sub-agent context for the subagent_ask WS route. + // Optional (subagent_ask returns a clear error when unset). + Scheduler SubagentAuditor + // Tasks queries task output for the task_output_query WS route. + // Optional (task_output_query returns a clear error when unset). + Tasks TaskOutputStore + // Streaming marks the injected Loop as streaming-capable (its OnStream is + // wired to publish ws.stream_chunk). When set the hub skips its own single + // final stream_chunk emission, since deltas already stream through OnStream. + Streaming bool // Frontend is the SPA filesystem served for all non-API, non-WS routes. // When nil the gateway serves no static assets (API and WS only). The // embedded dist (internal/webui) or an external dist directory is used. @@ -130,18 +154,21 @@ type Deps struct { // Server is the gateway. It owns a ServeMux aggregating REST + WS + admin and a // WebSocket hub. It is safe for concurrent use once constructed. type Server struct { - cfg *config.Config - auth Authenticator - bus bus.Bus - caps CapabilityStore - runner Chatter - sessions SessionStore - personas PersonaStore - memories MemoryStore - iot IoTStore - frontend fs.FS - log *slog.Logger - hub *hub + cfg *config.Config + auth Authenticator + bus bus.Bus + caps CapabilityStore + runner Chatter + sessions SessionStore + personas PersonaStore + memories MemoryStore + iot IoTStore + auditor SubagentAuditor + taskOutputs TaskOutputStore + streaming bool + frontend fs.FS + log *slog.Logger + hub *hub } // New validates cfg and deps and returns a ready Server. Auth is required; the @@ -155,17 +182,20 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { return nil, fmt.Errorf("gateway: auth dependency required") } s := &Server{ - cfg: cfg, - auth: deps.Auth, - bus: deps.Bus, - caps: deps.Registry, - runner: deps.Loop, - sessions: deps.Session, - personas: deps.Personas, - memories: deps.Memories, - iot: deps.IoT, - frontend: deps.Frontend, - log: sourcedLog(deps.Log), + cfg: cfg, + auth: deps.Auth, + bus: deps.Bus, + caps: deps.Registry, + runner: deps.Loop, + sessions: deps.Session, + personas: deps.Personas, + memories: deps.Memories, + iot: deps.IoT, + auditor: deps.Scheduler, + taskOutputs: deps.Tasks, + streaming: deps.Streaming, + frontend: deps.Frontend, + log: sourcedLog(deps.Log), } s.hub = newHub(s) s.hub.start() diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go new file mode 100644 index 0000000..8efa9ab --- /dev/null +++ b/internal/services/gateway/ws_bridge.go @@ -0,0 +1,92 @@ +package gateway + +import ( + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" +) + +// eventBridge maps a producer bus subject to its outbound WS event type. The hub +// subscribes to these (and re-normalizes into ws.* types) so the frontend's +// existing consumers light up without the producers changing their subjects. +var eventBridge = map[string]string{ + "iot.device_update": "device_update", + "task.output": "task_output", + "task.done": "task_done", + "automation.result": "automation_result", + "reminder.trigger": "reminder", +} + +// bridgePatterns are the additional bus patterns the hub subscribes to on top of +// ws.> to capture producer events and normalize them for WS delivery. +var bridgePatterns = []string{ + "iot.>", + "task.>", + "automation.>", + "reminder.>", + "cyrene.agent.*.event", +} + +// normalizeEvent maps a bus event to an outbound wsEvent. A ws.* subject keeps +// its design type (the ws. prefix is stripped); a known producer subject is +// bridged to its ws.* type; a sub-agent progress event is shaped into +// ws.subagent_event. It returns ok=false for events that have no WS-visible +// representation (e.g. internal cyrene.agent ask/reply traffic). +func normalizeEvent(e bus.Event) (wsEvent, bool) { + if typ, ok := eventBridge[e.Subject]; ok { + return wsEvent{Type: typ, Data: e.Payload}, true + } + if strings.HasPrefix(e.Subject, wsPrefix) { + return wsEvent{Type: eventType(e.Subject), Data: e.Payload}, true + } + if ev, ok := subagentProgressEvent(e); ok { + return ev, true + } + return wsEvent{}, false +} + +// subagentProgressEvent maps an orchestration progress event — broadcast on +// cyrene.agent..event via comm.PublishEvent — into a ws.subagent_event +// frame. Only the five sub-agent progress names are surfaced; any other event +// broadcast on the same subject (ask/reply/addressed events, or foreign event +// names) is ignored so internal agent-to-agent traffic stays internal. +func subagentProgressEvent(e bus.Event) (wsEvent, bool) { + if !isBroadcastAgentEvent(e.Subject) { + return wsEvent{}, false + } + ev, ok := e.Payload.(comm.Event) + if !ok { + return wsEvent{}, false + } + switch ev.Name { + case orchestration.EventDispatched, orchestration.EventStarted, + orchestration.EventDone, orchestration.EventFailed, orchestration.EventAborted: + default: + return wsEvent{}, false + } + p, ok := ev.Data.(orchestration.Progress) + if !ok { + return wsEvent{}, false + } + return wsEvent{Type: "subagent_event", Data: map[string]any{ + "id": p.AgentID, + "name": p.AgentID, + "status": string(p.Status), + "role": p.RoleID, + "mode": p.Mode, + }}, true +} + +// isBroadcastAgentEvent reports whether subject has the exact 4-token broadcast +// shape cyrene.agent..event, which is how the scheduler publishes progress. +// It deliberately does not match addressed events (cyrene.agent...event) +// or the ask/reply subjects. +func isBroadcastAgentEvent(subject string) bool { + parts := strings.Split(subject, ".") + if len(parts) != 4 { + return false + } + return parts[0] == "cyrene" && parts[1] == "agent" && parts[3] == "event" +} diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index 86c45f8..bd910d9 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -23,6 +23,7 @@ const ( wsEventsPattern = "ws.>" wsPrefix = "ws." wsChatSubject = "ws.chat" + wsVoiceInput = "ws.voice_input" wsReadLimit = 4096 wsWriteWait = 10 * time.Second wsPongWait = 60 * time.Second @@ -57,27 +58,33 @@ type wsConn struct { // hub owns the connection set and relays bus events to every client. type hub struct { - mu sync.RWMutex - conns map[*wsConn]struct{} - bus bus.Bus - runner Chatter - log *slog.Logger - upgrader websocket.Upgrader - origins map[string]bool - allowAll bool - cancel context.CancelFunc + mu sync.RWMutex + conns map[*wsConn]struct{} + bus bus.Bus + runner Chatter + auditor SubagentAuditor + taskOutputs TaskOutputStore + streaming bool + log *slog.Logger + upgrader websocket.Upgrader + origins map[string]bool + allowAll bool + cancel context.CancelFunc } // newHub builds a hub bound to the server's dependencies. func newHub(s *Server) *hub { origins, allowAll := s.allowedOrigins() return &hub{ - conns: make(map[*wsConn]struct{}), - bus: s.bus, - runner: s.runner, - log: s.log, - origins: origins, - allowAll: allowAll, + conns: make(map[*wsConn]struct{}), + bus: s.bus, + runner: s.runner, + auditor: s.auditor, + taskOutputs: s.taskOutputs, + streaming: s.streaming, + log: s.log, + origins: origins, + allowAll: allowAll, upgrader: websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, @@ -87,15 +94,20 @@ func newHub(s *Server) *hub { } // start subscribes the hub to the bus event stream. Publishing to ws.> arrives -// here and is fanned out to clients; nothing happens when no bus is wired. +// here and is fanned out to clients; the producer bridge patterns (iot/task/ +// automation/reminder/subagent-progress) are also subscribed and normalized into +// ws.* event types. Nothing happens when no bus is wired. func (h *hub) start() { if h.bus == nil { return } ctx, cancel := context.WithCancel(context.Background()) h.cancel = cancel - if _, err := h.bus.Subscribe(ctx, wsEventsPattern, h.relay); err != nil { - h.log.Warn("bus subscribe", "pattern", wsEventsPattern, "err", err) + patterns := append([]string{wsEventsPattern}, bridgePatterns...) + for _, p := range patterns { + if _, err := h.bus.Subscribe(ctx, p, h.relay); err != nil { + h.log.Warn("bus subscribe", "pattern", p, "err", err) + } } } @@ -154,15 +166,23 @@ func (h *hub) broadcast(ev wsEvent) { } } -// relay is the bus handler for the ws.> pattern. The ws.chat subject is a -// streaming trigger (run the loop and emit chunks); every other subject becomes -// an outbound event of its design type. +// relay is the bus handler for the ws.> pattern and the producer bridge +// patterns. The ws.chat subject is a streaming trigger (run the loop and emit +// chunks); a ws.* subject becomes an outbound event of its design type; a +// bridged producer subject (device_update, task output/done, automation result, +// reminder, subagent progress) is normalized into the matching ws.* event type. +// Events with no WS-visible mapping are dropped so internal agent-to-agent +// traffic never leaks to clients. func (h *hub) relay(ctx context.Context, e bus.Event) error { if e.Subject == wsChatSubject { go h.streamChat(ctx, e.Payload) return nil } - h.broadcast(wsEvent{Type: eventType(e.Subject), Data: e.Payload}) + ev, ok := normalizeEvent(e) + if !ok { + return nil + } + h.broadcast(ev) return nil } @@ -187,7 +207,13 @@ func (h *hub) streamChat(ctx context.Context, payload any) { h.broadcast(wsEvent{Type: "stream_end", Data: map[string]any{"stopReason": "error", "error": err.Error()}}) return } - h.broadcast(wsEvent{Type: "stream_chunk", Data: map[string]any{"delta": res.Text}}) + // When the loop is streaming (OnStream wired), the deltas already fan out as + // ws.stream_chunk through the OnStream→bus→relay path; a second whole-text + // chunk would duplicate the tail. The non-streaming path (or a stub runner) + // still emits one chunk so the stream contract holds in tests. + if !h.streaming { + h.broadcast(wsEvent{Type: "stream_chunk", Data: map[string]any{"delta": res.Text}}) + } h.broadcast(wsEvent{Type: "stream_end", Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}}) } @@ -282,22 +308,6 @@ func (c *wsConn) writePump() { } } -// route dispatches one inbound frame by its type. -func (c *wsConn) route(in wsInbound) { - switch in.Type { - case "mode_switch": - c.publishPublic("ws.mode_switch", rawData(in.Data)) - case "chat": - c.publishPublic(wsChatSubject, rawData(in.Data)) - case "subagent_ask": - c.enqueueJSON(wsEvent{Type: "subagent_event", Data: map[string]any{"state": "pending"}}) - case "task_output_query": - c.enqueueJSON(wsEvent{Type: "task_output", Data: map[string]any{}}) - default: - c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "unknown event type"}}) - } -} - // publishPublic publishes a bus event without any per-connection reply. func (c *wsConn) publishPublic(subject string, payload any) { if c.h.bus == nil { @@ -306,11 +316,14 @@ func (c *wsConn) publishPublic(subject string, payload any) { _ = c.h.bus.Publish(context.Background(), subject, payload) } -// shutdown closes the connection exactly once. +// shutdown closes the connection exactly once. It tolerates a nil *websocket.Conn +// so tests that build a fake connection can rely on a safe teardown. func (c *wsConn) shutdown() { c.once.Do(func() { close(c.done) - _ = c.conn.Close() + if c.conn != nil { + _ = c.conn.Close() + } }) } diff --git a/internal/services/gateway/ws_route.go b/internal/services/gateway/ws_route.go new file mode 100644 index 0000000..ea2bfbc --- /dev/null +++ b/internal/services/gateway/ws_route.go @@ -0,0 +1,107 @@ +package gateway + +import ( + "context" + "encoding/json" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" +) + +// route dispatches one inbound frame by its type. +func (c *wsConn) route(in wsInbound) { + switch in.Type { + case "mode_switch": + c.publishPublic("ws.mode_switch", rawData(in.Data)) + case "chat": + c.publishPublic(wsChatSubject, rawData(in.Data)) + case "voice_input": + // Voice is not wired to a live pipeline yet; surface the payload on the + // bus so an upper layer (or a later milestone) can consume it, rather + // than dropping it as unknown. + c.publishPublic(wsVoiceInput, rawData(in.Data)) + case "subagent_ask": + c.handleSubagentAsk(in) + case "task_output_query": + c.handleTaskOutputQuery(in) + default: + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "unknown event type"}}) + } +} + +// subagentAskRequest is the decoded payload of a subagent_ask frame. Mode selects +// the audit window (""/full/tail/keyword); n bounds a tail query; query is the +// keyword for a keyword query. +type subagentAskRequest struct { + ID string `json:"id"` + Mode string `json:"mode"` + N int `json:"n"` + Query string `json:"query"` +} + +// taskOutputRequest is the decoded payload of a task_output_query frame. +type taskOutputRequest struct { + TaskID string `json:"taskId"` +} + +// handleSubagentAsk audits a sub-agent's context and replies with a +// ws.subagent_event frame carrying the real context window. When no auditor is +// injected (gateway not wired with a scheduler) it replies with a clear error +// rather than a fabricated "pending" snapshot. +func (c *wsConn) handleSubagentAsk(in wsInbound) { + if c.h.auditor == nil { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "subagent audit unavailable"}}) + return + } + var req subagentAskRequest + _ = json.Unmarshal(in.Data, &req) + if req.ID == "" { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "subagent id required"}}) + return + } + mode := orchestration.AuditMode(req.Mode) + if mode == "" { + mode = orchestration.AuditFull + } + res, err := c.h.auditor.AuditContext(context.Background(), req.ID, mode, req.N, req.Query) + if err != nil { + c.h.log.Debug("subagent audit", "id", req.ID, "err", err) + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": err.Error()}}) + return + } + c.enqueueJSON(wsEvent{Type: "subagent_event", Data: res}) +} + +// handleTaskOutputQuery fetches the real task output for taskId and replies with +// a ws.task_output frame. When no task store is injected it replies with a clear +// error rather than an empty {}. A task that is still running returns the output +// captured so far. +func (c *wsConn) handleTaskOutputQuery(in wsInbound) { + if c.h.taskOutputs == nil { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "task output unavailable"}}) + return + } + var req taskOutputRequest + _ = json.Unmarshal(in.Data, &req) + if req.TaskID == "" { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "task id required"}}) + return + } + ev, err := c.h.taskOutputs.TaskOutput(context.Background(), req.TaskID) + if err != nil { + c.h.log.Debug("task output query", "id", req.TaskID, "err", err) + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": err.Error()}}) + return + } + text := ev.Output.Stdout + if ev.Output.Stderr != "" { + if text != "" { + text += "\n" + } + text += ev.Output.Stderr + } + c.enqueueJSON(wsEvent{Type: "task_output", Data: map[string]any{ + "taskId": ev.ID, + "output": text, + "status": string(ev.Status), + }}) +}