diff --git a/Docs/api-client.md b/Docs/api-client.md index ae8c714..2139941 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -21,7 +21,7 @@ |---|------|------|------|:---:|------| | 1 | `/api/v1/auth/login` | POST | 否 | ✅ | `{username,password}` → `{data:{token,jwt}}`(24h,owner) | | 2 | `/api/v1/health` | GET | 否 | ✅ | 健康检查 `{data:{status:"ok"}}` | -| 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,mode?,stream?}` → `{data:{reply}}`;`stream:true` 时经 WS 触发流式 | +| 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,stream?}` → `{data:{reply}}`;`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL,入视觉真推理(design/06) | | 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}` | | 5 | `/api/v1/personas` | GET | JWT | ✅ | `{data:[persona...]}` | @@ -38,17 +38,17 @@ | 5 | `/api/v1/admin/plugins/policy` | POST | ✅ | 写工具策略(owner-only,审计) | | 6 | `/api/v1/admin/channels` | GET | ✅ | 渠道列表/健康 | | 7 | `/api/v1/admin/iot` | GET | ✅ | IoT 设备列表 | -| 8 | `/api/v1/admin/users` | GET/POST | 🕐 | 用户/角色/API key(M10) | -| 9 | `/api/v1/admin/llm` | GET/PUT | 🕐 | LLM 模型/路由/降级(M10) | -| 10 | `/api/v1/admin/automations` | GET/POST | 🕐 | 自动化与提醒(M10) | +| 8 | `/api/v1/admin/users` | GET/POST | ✅(GET) | 用户/角色/API key(GET 已实现;POST 创建用户/签发 key 随 M10 落地) | +| 9 | `/api/v1/admin/llm` | GET/PUT | ✅(GET) | LLM 模型/路由/降级(GET 已实现,APIKey 脱敏;PUT 热重载 M10) | +| 10 | `/api/v1/admin/automations` | GET/POST | ✅(GET) | 自动化与提醒(GET 已实现;POST 创建 M10) | +| 11 | `/api/v1/admin/backthink` | GET | ✅ | 后台思考记录 | -> **M10 前端契约补充(管理面板,均🕐未实现,前端按契约封装、后端 404 时显示"待实现"壳,不报错)**: -> `GET /api/v1/admin/users`(用户/角色/API key)· `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` · `GET /api/v1/admin/logs` · `GET /api/v1/admin/trace` · `GET /api/v1/admin/audit` · `GET /api/v1/admin/system`。写操作(策略/档案/密钥/系统设置)经前端审计登记,待后端 M10+ 落地真实持久化与热重载。 +> **M10 前端契约补充(管理面板)**:`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` 已实现,返回真实数据;`GET /api/v1/admin/logs` · `GET /api/v1/admin/trace` · `GET /api/v1/admin/audit` · `GET /api/v1/admin/system` 仍🕐未实现(前端按契约封装、后端 404 时显示"待实现"壳,不报错)。写操作(策略/档案/密钥/系统设置/用户与自动化创建)经前端审计登记,待后端 M10+ 落地真实持久化与热重载。 ## 3. WebSocket `/ws/chat`(实时) - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`;未认证在握手前回 `401` JSON。 -- **下行(服务端→客户端)**: 经 bus 广播 `{type:"", data:..., turnId?, seq?}`(`turnId`/`seq` 为流式关联字段,见 §3.1)——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`(结构化流,design/04 §1.1),以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood` / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update`。 +- **下行(服务端→客户端)**: 经 bus 广播 `{type:"", data:..., turnId?, seq?}`(`turnId`/`seq` 为流式关联字段,见 §3.1)——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`(结构化流,design/04 §1.1),以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood` / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update`。语音回复走 `voice_output`(定向回请求连接,非广播)。 ### 3.1 结构化流事件(`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*`) - **事件壳**:`{type, turnId?, seq?, data}`, `data` 为**类型化结构**(非 `any`)。`seq` 为 hub 级单调递增序号,供客户端将一次突发的流式事件排序关联。 @@ -58,7 +58,9 @@ - **`tool_execution.update`**(`{id, index, delta}`):工具执行进度/局部输出(保留,工具循环 §1.2 后置实现)。 - **向后兼容**:未映射/无法确定的事件仍走 `stream_chunk` + `stream_end`(非流式回退路径)。 - **上行(客户端→服务端)**: `mode_switch` / `chat`(跑 loop 并广播流式) / `voice_input` / `subagent_ask` / `task_output_query`。 -- ⚠️ 语音/流式完整 schema 随 M9 前端联调补充;会话级定向(当前对所有连接广播)为后续项。 +- **`voice_input`**(上行,客户端→服务端):`_{audio?, format?, text?, sessionId?}`。`audio` 为 base64(或 data: URI)音频,`format` 为其编码(`wav`/`mp3`/`ogg`);`text` 为已转写文本时跳过 STT。服务端走 STT→agent 聊天→TTS(design/06 §1),回复经 `voice_output` 定向回该连接。无注入语音服务时回 `error` 帧(`voice unavailable`)。 +- **`voice_output`**(下行,服务端→客户端):`{audio, format, text, reply}`。`audio` 为 base64 合成音频,`format` 为其编码,`text` 为识别文本,`reply` 为助手回复文本。 +- **向后兼容**:未映射/无法确定的事件仍走 `stream_chunk` + `stream_end`(非流式回退路径)。 --- @@ -68,6 +70,7 @@ - **离线**: PWA / 原生端建议缓存 `sessions`/`personas` 列表。 ## 5. 变更记录 +- `v0.5`(批次B·视觉真推理 + 语音端):`POST /api/v1/chat` 增 `image` 字段(base64 data: URI 或 URL),入 loop 视觉路由(vision/OCR 工具或主模型多模态,design/06 §0「先看图再对话」);WS 上行 `voice_input` 接 STT→聊天→TTS 管道,下行新增 `voice_output`(定向回请求连接,含合成音频 + 识别文本 + 回复文本)。 - `v0.4`(结构化流):WS 下行新增 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 结构化流事件(event shell 补 `turnId`/`seq`,`data` 类型化),对齐 OpenClaw AssistantMessageEvent;保留 `stream_chunk`/`stream_end` 向后兼容。 - `v0.3`(M10):管理面板 `/admin` 落地——按 §2 契约封装 admin 只读端点(personas/sessions/memories/plugins/channels/iot)+ owner-only `plugins/policy` 写;其余 M10 端点以契约路径封装、后端未实现时前端显式"待实现"壳。 - `v0.2`(M8):gateway 落地——登录/health/chat/sessions/personas + admin 只读端点 + `plugins/policy` 写 + WS `/ws/chat`(JWT + bus 广播)。 diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 6209c76..2cd41b8 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -31,6 +31,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/vision" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/audit" @@ -297,6 +298,21 @@ func (a *app) wireRuntime() error { return fmt.Errorf("checkpoint store: %w", perr) } + // Vision routing: the loop rewrites an image-bearing user message per the + // vision Router — a vision-capable main model sees the image directly, + // otherwise the vision/OCR tool's text description replaces it (design/06 §0). + visionRouter := vision.NewRouter( + vision.WithVisionTool(visionToolName), + vision.WithOCRTool(ocrToolName), + ) + visionResolver := vision.NewMessageResolver( + visionRouter, + func() vision.ModelInfo { + return vision.ModelInfo{Name: a.cfg.LLM.VisionModel, SupportsVision: a.cfg.LLM.VisionModel != ""} + }, + a.dispatcher.Invoke, + ) + a.loop = loop.New(loop.Options{ Agent: ag, Tools: a.dispatcher, @@ -310,6 +326,7 @@ func (a *app) wireRuntime() error { MaxRounds: defaultMaxRounds, WorkingMemory: memory.NewWorkingMemory(0), OnStream: a.streamChunk, + Vision: visionResolver, Log: logger.WithSource(a.log, "runtime/loop"), }) @@ -417,24 +434,29 @@ func (a *app) wireChannels() error { // managers so the admin/user list endpoints return real data. func (a *app) wireFrontend() error { jwt := auth.New(a.cfg.Auth) + apiKeys := auth.NewAPIKey(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, - Scheduler: a.scheduler, - Tasks: a.tasks, - Streaming: true, - 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, + Automations: a.automation, + Users: apiKeys, + Backthink: a.backthink, + Streaming: true, + Frontend: resolveFrontendFS(a.log), }) if err != nil { return fmt.Errorf("gateway: %w", err) } + gw.SetVoice(a.voice) a.gw = gw return nil } diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index f4d2d5b..d702eb5 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -37,6 +37,34 @@ function toRecord(v: unknown): Record | null { return null; } +/** Decode a base64 string into bytes (used for the voice audio reply). */ +function decodeBase64(b64: string): Uint8Array { + const bin = atob(b64); + const bytes = new Uint8Array(bin.length); + for (let i = 0; i < bin.length; i++) bytes[i] = bin.charCodeAt(i); + return bytes; +} + +/** Best-effort playback of a voice_output audio payload (blocked by autoplay + * policy, so sound may be silent until the user interacts). */ +function playVoiceAudio(payload: unknown): void { + const r = toRecord(payload); + if (!r || typeof r.audio !== "string" || !r.audio) return; + try { + const bytes = decodeBase64(r.audio); + const format = typeof r.format === "string" && r.format ? r.format : "wav"; + const buf = new ArrayBuffer(bytes.byteLength); + new Uint8Array(buf).set(bytes); + const blob = new Blob([buf], { type: `audio/${format}` }); + const url = URL.createObjectURL(blob); + const audio = new Audio(url); + void audio.play().catch(() => {}); + audio.onended = () => URL.revokeObjectURL(url); + } catch { + // Ignore malformed audio payloads. + } +} + function parseTool(v: unknown): ToolUsed | null { const r = toRecord(v); if (!r) return null; @@ -253,6 +281,18 @@ function dispatch(ev: { type: string; data: unknown }): void { } break; } + case "voice_output": { + // Voice turn reply: the backend sends back the synthesized audio plus the + // recognized transcript and the assistant reply. Playback is best-effort + // (autoplay may be blocked); the transcript/reply is surfaced as a log line. + playVoiceAudio(ev.data); + if (r && typeof r.reply === "string") { + s.addDeviceUpdate(`语音回复:${r.reply}`); + } else if (r && typeof r.text === "string") { + s.addDeviceUpdate(`语音识别:${r.text}`); + } + break; + } default: { // Unknown future event: keep it in the context log so no frame is lost. const text = typeof ev.data === "string" ? ev.data : JSON.stringify(ev.data); diff --git a/internal/auth/apikey.go b/internal/auth/apikey.go index 06a53ac..c656ba0 100644 --- a/internal/auth/apikey.go +++ b/internal/auth/apikey.go @@ -71,3 +71,26 @@ func (k *APIKey) Revoke(key string) { defer k.mu.Unlock() delete(k.keys, key) } + +// List returns a summary of every currently registered API key for the admin +// panel. Key material is never returned; each identifier is masked so the panel +// can tell keys apart without exposing the secret. The in-memory store keeps no +// creation time, so CreatedAt is nil for every entry. +func (k *APIKey) List() []UserSummary { + k.mu.RLock() + defer k.mu.RUnlock() + out := make([]UserSummary, 0, len(k.keys)) + for key, role := range k.keys { + out = append(out, UserSummary{ID: maskKey(key), Username: "apikey", Role: role}) + } + return out +} + +// maskKey renders a key's leading and trailing groups so the panel can identify +// an entry without revealing the secret. Short keys are fully hidden. +func maskKey(key string) string { + if len(key) <= 8 { + return "••••" + } + return key[:4] + "••••" + key[len(key)-4:] +} diff --git a/internal/auth/auth.go b/internal/auth/auth.go index 8d272c0..b88ff67 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -12,6 +12,7 @@ package auth import ( "errors" "fmt" + "time" ) // Role is an authorization role in the owner>admin>viewer hierarchy. @@ -91,3 +92,14 @@ func (r Role) rank() int { func HasRole(role, required Role) bool { return role.rank() >= required.rank() } + +// UserSummary is the read-only shape of an account or API key visible to the +// admin panel (GET /api/v1/admin/users). CreatedAt is nil when the source does +// not track a creation time (the configured admin account and in-memory keys +// both omit it). +type UserSummary struct { + ID string + Username string + Role Role + CreatedAt *time.Time +} diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go index 95d6286..02d2a07 100644 --- a/internal/runtime/loop/loop.go +++ b/internal/runtime/loop/loop.go @@ -97,6 +97,7 @@ type Loop struct { maxContextTokens int compactor Compactor transformContext func(context.Context, []model.Message, int) []model.Message + imageResolver ImageResolver log *slog.Logger } @@ -141,6 +142,7 @@ func New(opts Options) *Loop { maxContextTokens: opts.MaxContextTokens, compactor: compactor, transformContext: opts.TransformContext, + imageResolver: opts.Vision, log: sourcedLog(opts.Log), } } diff --git a/internal/runtime/loop/loop_vision_test.go b/internal/runtime/loop/loop_vision_test.go new file mode 100644 index 0000000..e54067c --- /dev/null +++ b/internal/runtime/loop/loop_vision_test.go @@ -0,0 +1,65 @@ +package loop + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +func TestRun_ImageResolverFoldsToText(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Model: "m", Content: "Done."}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{ + Agent: a, + MaxRounds: 8, + Vision: ImageResolveFunc(func(_ context.Context, _ model.Message) (model.Message, error) { + return model.TextMessage(model.RoleUser, "图中是一只猫"), nil + }), + }) + + in := []model.Message{model.ImageMessage(model.RoleUser, "", "data:image/png;base64,AAAA")} + if _, err := l.Run(context.Background(), "sess-1", in); err != nil { + t.Fatalf("Run: %v", err) + } + if len(ad.reqs) == 0 { + t.Fatalf("no model request recorded") + } + var userSeen bool + for _, m := range ad.reqs[0].Messages { + if m.Role == model.RoleUser { + userSeen = true + if messageHasImage(m) { + t.Fatalf("image still reached the model: %+v", m) + } + } + } + if !userSeen { + t.Fatalf("no user message seen by the model") + } +} + +func TestRun_ImageKeptWithoutResolver(t *testing.T) { + t.Parallel() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Model: "m", Content: "Done."}}}} + a := newTestAgent(t, ad, nil) + l := New(Options{Agent: a, MaxRounds: 8}) + + in := []model.Message{model.ImageMessage(model.RoleUser, "look", "data:image/png;base64,AAAA")} + if _, err := l.Run(context.Background(), "sess-1", in); err != nil { + t.Fatalf("Run: %v", err) + } + if len(ad.reqs) == 0 { + t.Fatalf("no model request recorded") + } + var sawImage bool + for _, m := range ad.reqs[0].Messages { + if m.Role == model.RoleUser && messageHasImage(m) { + sawImage = true + } + } + if !sawImage { + t.Fatalf("expected the image part to pass through without a resolver") + } +} diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index 492507d..8ada3ea 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -77,6 +77,13 @@ type Options struct { // an LLM-backed or custom summarizer. It is only invoked when // MaxContextTokens is positive. Compactor Compactor + // Vision, when non-nil, rewrites an image-bearing user message at the start + // of Run so the main model can reason over the image (design/06 §0). The + // concrete resolver (vision.MessageResolver) decides via the vision Router + // whether to keep the image for a vision-capable main model or fold it into a + // text description from the vision/OCR tool ("先看图再对话"). Nil keeps the + // image content as-is, passing it through to a vision-capable main model. + Vision ImageResolver // TransformContext, when non-nil, is called immediately before every model // step with the assembled message list (after memory injection and any // compaction) and the current round number (1-based). It may project, filter diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 584474f..3059cdb 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -31,7 +31,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa return nil, fmt.Errorf("loop: run: %w", ErrNilAgent) } - msgs := messages + msgs := l.resolveImages(ctx, messages) toolState := &toolLoopState{prevFP: make(map[string]bool)} res := &Result{} for step := 1; step <= l.maxRounds; step++ { diff --git a/internal/runtime/loop/vision.go b/internal/runtime/loop/vision.go new file mode 100644 index 0000000..bfd716f --- /dev/null +++ b/internal/runtime/loop/vision.go @@ -0,0 +1,71 @@ +package loop + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// ImageResolver is the loop's seam to the multimodal vision layer (design/06 +// §0). It is called once at the start of Run for every message that carries an +// image content part, and may rewrite that message so the main model can reason +// over the image. The concrete resolver lives in the vision package and decides, +// via the vision Router, whether to keep the image for a vision-capable main +// model or fold it into a text description produced by the vision/OCR tool ("先 +// 看图再对话"). Nil leaves the image content untouched so it flows straight to a +// vision-capable main model. +type ImageResolver interface { + // Resolve rewrites msg, which carries at least one image content part. It + // returns the (possibly rewritten) message. An error leaves the original + // message in place so the turn still proceeds. + Resolve(ctx context.Context, msg model.Message) (model.Message, error) +} + +// ImageResolveFunc adapts a function to the ImageResolver interface. +type ImageResolveFunc func(ctx context.Context, msg model.Message) (model.Message, error) + +// Resolve implements ImageResolver. +func (f ImageResolveFunc) Resolve(ctx context.Context, msg model.Message) (model.Message, error) { + return f(ctx, msg) +} + +// resolveImages rewrites every image-bearing message in messages using the +// injected resolver. It returns the input slice unchanged when no resolver is +// configured, no image is present, or a message fails to resolve (the failure is +// logged and the original message is kept so the turn still runs). A fresh slice +// is returned when any message changed. +func (l *Loop) resolveImages(ctx context.Context, messages []model.Message) []model.Message { + if l.imageResolver == nil { + return messages + } + out := make([]model.Message, len(messages)) + changed := false + for i, m := range messages { + if !messageHasImage(m) { + out[i] = m + continue + } + resolved, err := l.imageResolver.Resolve(ctx, m) + if err != nil { + l.log.Warn("image resolve", "error", err) + out[i] = m + continue + } + out[i] = resolved + changed = true + } + if !changed { + return messages + } + return out +} + +// messageHasImage reports whether any content part of msg is an image part. +func messageHasImage(m model.Message) bool { + for _, part := range m.Content { + if part.Type == model.ContentTypeImage { + return true + } + } + return false +} diff --git a/internal/runtime/multimodal/vision/resolver.go b/internal/runtime/multimodal/vision/resolver.go new file mode 100644 index 0000000..ac1020a --- /dev/null +++ b/internal/runtime/multimodal/vision/resolver.go @@ -0,0 +1,123 @@ +package vision + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// ModelInfoProvider supplies the main model's vision capability so the resolver +// can route an image the way the vision Router decides (design/06 §0). It is +// called per image; the same provider is shared across a single resolution. +type ModelInfoProvider func() ModelInfo + +// ToolInvoker invokes a registered tool (the vision or OCR tool) by name with +// the given arguments. It is satisfied by *tool.Dispatcher.Invoke. +type ToolInvoker func(ctx context.Context, name string, args map[string]any) (any, error) + +// MessageResolver is the loop's concrete ImageResolver. For each image part in a +// user message it asks the vision Router how to route it: when the main model +// supports vision and is confident, the image part is kept so the main model sees +// the image; otherwise the vision/OCR tool is invoked and its text result replaces +// the image part, so a text-only main model can reason over the image ("先看图再 +// 对话"). It is safe for concurrent use. +type MessageResolver struct { + router *Router + info ModelInfoProvider + invoke ToolInvoker +} + +// NewMessageResolver builds a MessageResolver. router decides the layer; info +// reports the main model's vision capability; invoke runs a registered vision/OCR +// tool. A nil invoke leaves any tool route failed (the image is then kept), so a +// resolver without tools degrades to the main-model path. +func NewMessageResolver(router *Router, info ModelInfoProvider, invoke ToolInvoker) *MessageResolver { + return &MessageResolver{ + router: router, + info: info, + invoke: invoke, + } +} + +// Resolve rewrites msg so every image part is either kept (main model sees it) or +// replaced by a text description produced by the vision/OCR tool. Text parts are +// preserved in order. It errors only when an image cannot be routed or a tool +// invocation fails; the caller (loop) then keeps the original message. +func (r *MessageResolver) Resolve(ctx context.Context, msg model.Message) (model.Message, error) { + if r.router == nil { + return msg, nil + } + out := make([]model.Content, 0, len(msg.Content)) + for _, part := range msg.Content { + if part.Type != model.ContentTypeImage { + out = append(out, part) + continue + } + replacement, err := r.resolveImage(ctx, part.ImageURL) + if err != nil { + return model.Message{}, err + } + out = append(out, replacement) + } + return model.Message{Role: msg.Role, Content: out}, nil +} + +// resolveImage picks a route and either keeps the image (KindMain) or folds it +// into a text description via the vision/OCR tool. +func (r *MessageResolver) resolveImage(ctx context.Context, imageURL string) (model.Content, error) { + target := ModelInfo{} + if r.info != nil { + target = r.info() + } + decision := r.router.Route(target, RouteRequest{}) + switch decision.Kind { + case KindMain: + // A vision-capable main model sees the image directly. + return model.Content{Type: model.ContentTypeImage, ImageURL: imageURL}, nil + case KindVisionTool: + return r.invokeTool(ctx, decision.Tool, map[string]any{ + "image": imageURL, + "prompt": "Describe this image in detail.", + }, "description") + case KindOCRTool: + return r.invokeTool(ctx, decision.Tool, map[string]any{ + "image": imageURL, + }, "text") + default: + return model.Content{}, fmt.Errorf("image route: %w: %s", ErrUnsupportedSource, decision.Reason) + } +} + +// invokeTool runs a registered vision/OCR tool and turns its text result into a +// text content part. field is the result map key ("description" or "text"). +func (r *MessageResolver) invokeTool(ctx context.Context, toolName string, args map[string]any, field string) (model.Content, error) { + if r.invoke == nil { + return model.Content{}, fmt.Errorf("%s: %w", toolName, ErrNotConfigured) + } + result, err := r.invoke(ctx, toolName, args) + if err != nil { + return model.Content{}, fmt.Errorf("image tool %s: %w", toolName, err) + } + return model.Content{Type: model.ContentTypeText, Text: descriptionFragment(fieldText(result, field))}, nil +} + +// fieldText extracts a string field from a tool result map, tolerating the +// map[string]any shape returned by the vision/OCR tools. +func fieldText(result any, field string) string { + m, ok := result.(map[string]any) + if !ok { + return "" + } + s, _ := m[field].(string) + return s +} + +// descriptionFragment labels a tool-produced description so the model can tell it +// refers to the picture the user attached. +func descriptionFragment(desc string) string { + if desc == "" { + return "[图片内容] 无法识别该图片" + } + return "[图片内容] " + desc +} diff --git a/internal/runtime/multimodal/vision/resolver_test.go b/internal/runtime/multimodal/vision/resolver_test.go new file mode 100644 index 0000000..4107080 --- /dev/null +++ b/internal/runtime/multimodal/vision/resolver_test.go @@ -0,0 +1,94 @@ +package vision + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +func TestMessageResolver_FoldsImageToDescription(t *testing.T) { + t.Parallel() + rt := NewRouter(WithVisionTool("vision"), WithOCRTool("ocr")) + invoked := "" + res := NewMessageResolver(rt, func() ModelInfo { + return ModelInfo{Name: "text-model", SupportsVision: false} + }, func(_ context.Context, name string, args map[string]any) (any, error) { + invoked = name + return map[string]any{"description": "一只橘色的猫"}, nil + }) + + msg, err := res.Resolve(context.Background(), model.ImageMessage(model.RoleUser, "看到了吗", "data:image/png;base64,AAAA")) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if invoked != "vision" { + t.Fatalf("invoked tool = %q, want vision", invoked) + } + var text string + for _, p := range msg.Content { + if p.Type == model.ContentTypeText { + text += p.Text + } + if p.Type == model.ContentTypeImage { + t.Fatalf("image part still present after tool route: %+v", p) + } + } + if text == "" || text == "看到了吗" { + t.Fatalf("expected a description fragment, got %q", text) + } +} + +func TestMessageResolver_KeepsImageForVisionModel(t *testing.T) { + t.Parallel() + rt := NewRouter(WithVisionTool("vision"), WithOCRTool("ocr")) + res := NewMessageResolver(rt, func() ModelInfo { + return ModelInfo{Name: "vision-model", SupportsVision: true} + }, func(_ context.Context, _ string, _ map[string]any) (any, error) { + t.Fatalf("tool should not be invoked when the main model is sighted") + return nil, nil + }) + + msg, err := res.Resolve(context.Background(), model.ImageMessage(model.RoleUser, "x", "data:image/png;base64,AAAA")) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if !messageHasImagePart(msg) { + t.Fatalf("expected the image part to be kept for a vision-capable main model") + } +} + +func TestMessageResolver_OCRToolForText(t *testing.T) { + t.Parallel() + rt := NewRouter(WithVisionTool("vision"), WithOCRTool("ocr")) + invoked := "" + res := NewMessageResolver(rt, func() ModelInfo { + return ModelInfo{Name: "text-model", SupportsVision: false} + }, func(_ context.Context, name string, _ map[string]any) (any, error) { + invoked = name + return map[string]any{"text": "HELLO"}, nil + }) + + // Force the text-primary hint by using a router route request path: this is + // exercised directly through Resolve which passes no hint, so the vision tool + // is preferred; assert at least a text part replaces the image. + msg, err := res.Resolve(context.Background(), model.ImageMessage(model.RoleUser, "x", "data:image/png;base64,AAAA")) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if invoked != "vision" { + t.Fatalf("invoked tool = %q, want vision (router prefers vision when no text hint)", invoked) + } + if messageHasImagePart(msg) { + t.Fatalf("expected the image to be replaced by text") + } +} + +func messageHasImagePart(m model.Message) bool { + for _, p := range m.Content { + if p.Type == model.ContentTypeImage { + return true + } + } + return false +} diff --git a/internal/services/automations/automations.go b/internal/services/automations/automations.go index c0dbd70..2c2b508 100644 --- a/internal/services/automations/automations.go +++ b/internal/services/automations/automations.go @@ -140,6 +140,20 @@ func (d Deps) pipeline() Pipeline { return nil } +// AutomationRecord is the read-only summary of one automation or reminder, +// returned by Service.List for the admin panel. Kind is "cron", "event" or +// "reminder"; Cron carries the cron expression for the cron kind; LastRun and +// NextRun are zero times when the value is unknown (typically before Start). +type AutomationRecord struct { + ID string + Name string + Kind string + Cron string + Enabled bool + LastRun time.Time + NextRun time.Time +} + // Result is the structured outcome of an executed automation, published to // SubjectResult. Upper layers subscribe to it to react or push notifications. type Result struct { diff --git a/internal/services/automations/service.go b/internal/services/automations/service.go index 7305a1f..2b253ab 100644 --- a/internal/services/automations/service.go +++ b/internal/services/automations/service.go @@ -22,6 +22,7 @@ type Service struct { mu sync.RWMutex cronRules map[string]ScheduleID // rule id -> schedule id + cronMeta map[string]Rule // rule id -> cron rule (name/expression for listing) } // NewService builds an automation engine from cfg and deps. The bus in deps is @@ -50,6 +51,7 @@ func NewService(cfg Config, deps Deps) (*Service, error) { engine: engine, remind: remind, cronRules: make(map[string]ScheduleID), + cronMeta: make(map[string]Rule), }, nil } @@ -87,6 +89,7 @@ func (s *Service) AddRule(ctx context.Context, r Rule) error { } s.mu.Lock() s.cronRules[r.ID] = id + s.cronMeta[r.ID] = r s.mu.Unlock() return nil default: @@ -100,6 +103,7 @@ func (s *Service) RemoveRule(ctx context.Context, id string) error { s.mu.Lock() if schedID, ok := s.cronRules[id]; ok { delete(s.cronRules, id) + delete(s.cronMeta, id) s.mu.Unlock() s.sched.RemoveSchedule(schedID) return nil @@ -116,6 +120,51 @@ func (s *Service) AddReminder(ctx context.Context, r Reminder) error { // Reminders returns a sorted snapshot of registered reminders. func (s *Service) Reminders() []Reminder { return s.remind.List() } +// List returns a read-only snapshot of every registered automation and reminder +// for the admin panel, composing the reminder store, the event rules and the +// cron schedules. It never mutates state and is safe for concurrent use. Cron +// entries' LastRun/NextRun are zero until the scheduler has run. +func (s *Service) List() []AutomationRecord { + out := make([]AutomationRecord, 0) + + for _, r := range s.remind.List() { + at := r.At + out = append(out, AutomationRecord{ + ID: r.ID, + Name: r.Message, + Kind: "reminder", + Enabled: true, + NextRun: at, + }) + } + + for _, r := range s.engine.Rules() { + out = append(out, AutomationRecord{ + ID: r.ID, + Name: r.Name, + Kind: string(r.Trigger), + Enabled: true, + }) + } + + s.mu.RLock() + for id, rule := range s.cronMeta { + entry := s.sched.Entry(s.cronRules[id]) + out = append(out, AutomationRecord{ + ID: id, + Name: rule.Name, + Kind: string(rule.Trigger), + Cron: rule.Expression, + Enabled: true, + LastRun: entry.Prev, + NextRun: entry.Next, + }) + } + s.mu.RUnlock() + + return out +} + // CancelReminder delegates to the ReminderStore. func (s *Service) CancelReminder(ctx context.Context, id string) error { return s.remind.Cancel(ctx, id) diff --git a/internal/services/gateway/admin.go b/internal/services/gateway/admin.go index 85bd2c6..0476dfa 100644 --- a/internal/services/gateway/admin.go +++ b/internal/services/gateway/admin.go @@ -5,12 +5,19 @@ import ( "encoding/json" "errors" "net/http" + "strconv" + "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" ) +// backthinkListLimit is the number of most-recent think records the admin +// panel asks for when no limit is specified. +const backthinkListLimit = 50 + // PluginSummary is the wire shape of a capability/plugin in the admin list. type PluginSummary struct { Name string `json:"name"` @@ -33,6 +40,55 @@ type policyRequest struct { AllowInStranger bool `json:"allowInStranger"` } +// AdminLlmProvider is the wire shape of one LLM provider in the admin config. +// The API key is deliberately absent: it is never returned to the client. +type AdminLlmProvider struct { + ID string `json:"id"` + Name string `json:"name"` + BaseURL string `json:"baseUrl"` + Model string `json:"model"` + Multimodal bool `json:"multimodal"` + Enabled bool `json:"enabled"` + Fallback string `json:"fallback,omitempty"` +} + +// AdminLlmConfig is the wire shape of GET /api/v1/admin/llm. +type AdminLlmConfig struct { + Providers []AdminLlmProvider `json:"providers"` +} + +// AdminAutomationRecord is the wire shape of one automation/reminder in the +// admin list. Cron carries the cron expression for the cron kind; lastRun and +// nextRun are omitted when the value is unknown. +type AdminAutomationRecord struct { + ID string `json:"id"` + Name string `json:"name"` + Kind string `json:"kind"` + Cron string `json:"cron,omitempty"` + Enabled bool `json:"enabled"` + LastRun string `json:"lastRun,omitempty"` + NextRun string `json:"nextRun,omitempty"` +} + +// AdminUserRecord is the wire shape of one account/API key in the admin list. +type AdminUserRecord struct { + ID string `json:"id"` + Username string `json:"username"` + Role string `json:"role"` + CreatedAt string `json:"createdAt,omitempty"` +} + +// AdminBackthinkItem is the wire shape of one background-think record in the +// admin monitor list. +type AdminBackthinkItem struct { + ID string `json:"id"` + Trigger string `json:"trigger"` + Conclusion string `json:"conclusion,omitempty"` + ShouldNotify bool `json:"shouldNotify"` + Pushed bool `json:"pushed"` + At int64 `json:"at"` +} + // adminListPersonas lists persona summaries (admin read). func (s *Server) adminListPersonas(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, s.collectPersonas()) @@ -97,6 +153,104 @@ func (s *Server) adminListIoT(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, out) } +// adminGetLlm returns the LLM provider/model configuration read from the live +// config. API keys are never returned (the wire shape carries no key field), so +// the handler leaks nothing sensitive. +func (s *Server) adminGetLlm(w http.ResponseWriter, r *http.Request) { + cfg := s.cfg.LLM + writeJSON(w, http.StatusOK, AdminLlmConfig{Providers: []AdminLlmProvider{ + { + ID: "openai", + Name: "OpenAI", + BaseURL: cfg.BaseURL, + Model: cfg.Model, + Multimodal: cfg.VisionModel != "", + Enabled: cfg.Provider != "anthropic", + Fallback: cfg.Fallback, + }, + { + ID: "anthropic", + Name: "Anthropic", + BaseURL: cfg.AnthropicBaseURL, + Model: cfg.AnthropicModel, + Multimodal: false, + Enabled: cfg.Provider == "anthropic", + Fallback: cfg.AnthropicFallback, + }, + }}) +} + +// adminListAutomations lists automations and reminders (admin read). +func (s *Server) adminListAutomations(w http.ResponseWriter, r *http.Request) { + out := make([]AdminAutomationRecord, 0) + if s.automations != nil { + for _, a := range s.automations.List() { + rec := AdminAutomationRecord{ + ID: a.ID, + Name: a.Name, + Kind: a.Kind, + Cron: a.Cron, + Enabled: a.Enabled, + } + if !a.LastRun.IsZero() { + rec.LastRun = a.LastRun.Format(time.RFC3339) + } + if !a.NextRun.IsZero() { + rec.NextRun = a.NextRun.Format(time.RFC3339) + } + out = append(out, rec) + } + } + writeJSON(w, http.StatusOK, out) +} + +// adminListUsers lists accounts and API keys (admin read). Without a wired +// UserStore it reports the single configured admin account rather than +// fabricating a column of virtual users. +func (s *Server) adminListUsers(w http.ResponseWriter, r *http.Request) { + out := []AdminUserRecord{{ + ID: "admin", + Username: s.cfg.Auth.AdminUser, + Role: string(auth.RoleOwner), + }} + if s.users != nil { + for _, u := range s.users.List() { + rec := AdminUserRecord{ID: u.ID, Username: u.Username, Role: string(u.Role)} + if u.CreatedAt != nil { + rec.CreatedAt = u.CreatedAt.Format(time.RFC3339) + } + out = append(out, rec) + } + } + writeJSON(w, http.StatusOK, out) +} + +// adminListBackthink lists the most recent background-think records (admin read). +func (s *Server) adminListBackthink(w http.ResponseWriter, r *http.Request) { + out := make([]AdminBackthinkItem, 0) + if s.backthink == nil { + writeJSON(w, http.StatusOK, out) + return + } + records, err := s.backthink.List(r.Context(), backthinkListLimit) + if err != nil { + s.log.Error("list backthink", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "backthink_failed", "failed to list backthink records")) + return + } + for _, rec := range records { + out = append(out, AdminBackthinkItem{ + ID: strconv.FormatInt(rec.ID, 10), + Trigger: rec.Trigger, + Conclusion: rec.Conclusion, + ShouldNotify: rec.ShouldPush, + Pushed: rec.PushResult.Pushed, + At: rec.At.UnixMilli(), + }) + } + writeJSON(w, http.StatusOK, out) +} + // adminSetPolicy upserts the reachability markers for a capability. It is the // audited admin write example (owner-only, enforced by the route registration). func (s *Server) adminSetPolicy(w http.ResponseWriter, r *http.Request) { diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 6d61d3d..0324616 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -25,6 +25,7 @@ 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/multimodal/voice" "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" @@ -918,10 +919,10 @@ func TestHubRoute(t *testing.T) { } 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). + // voice_input with no voice service → clear error, never a crash or a drop. 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) + if ev := readFakeFrame(t, c); ev.Type != "error" { + t.Fatalf("no-voice voice_input frame type = %q, want error", ev.Type) } // unknown → error (enqueueJSON). c.route(wsInbound{Type: "nope"}) @@ -935,6 +936,43 @@ func TestHubRoute(t *testing.T) { } } +// stubVoice is a deterministic VoiceService for voice route tests. +type stubVoice struct { + sttText string + sttErr error + tts voice.Audio + ttsErr error +} + +func (s *stubVoice) STT(_ context.Context, _ voice.Audio) (string, error) { return s.sttText, s.sttErr } +func (s *stubVoice) TTS(_ context.Context, _ string) (voice.Audio, error) { return s.tts, s.ttsErr } + +func TestHubVoiceInputTurn(t *testing.T) { + t.Parallel() + hb := newTestHub(&stubRunner{res: &loop.Result{Text: "你好", Steps: 1, StopReason: "stop"}}, nil) + hb.voice = &stubVoice{sttText: "你好", tts: voice.Audio{Format: "wav", Data: []byte("audio-bytes")}} + c := newFakeConn(hb, 32) + hb.register(c) + + // "aGVsbG8=" decodes to "hello" — a valid base64 audio payload. + c.route(wsInbound{Type: "voice_input", Data: json.RawMessage(`{"audio":"aGVsbG8=","format":"wav"}`)}) + // runVoiceTurn runs in a goroutine; read with a deadline. + ev := readFakeFrame(t, c) + if ev.Type != "voice_output" { + t.Fatalf("voice frame type = %q, want voice_output (frame=%+v)", ev.Type, ev) + } + r, ok := ev.Data.(map[string]any) + if !ok || r["text"] != "你好" { + t.Fatalf("voice transcript = %+v, want 你好", ev.Data) + } + if r["reply"] != "你好" { + t.Fatalf("voice reply = %+v, want 你好", ev.Data) + } + if r["audio"] == "" || r["audio"] == nil { + t.Fatalf("voice audio missing from reply: %+v", ev.Data) + } +} + func TestHubBridgeDeviceUpdate(t *testing.T) { t.Parallel() b := bus.New() diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index 3033c03..4d72499 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -57,10 +57,13 @@ type loginResponse struct { } // chatRequest is the POST /api/v1/chat body. When Stream is true the gateway -// triggers streaming over the WebSocket channel and returns immediately. +// triggers streaming over the WebSocket channel and returns immediately. Image is +// an optional base64 data: URI or http(s) URL attached for the vision path; the +// loop routes it through the vision/OCR tool or the main model (design/06 §0). type chatRequest struct { SessionID string `json:"sessionId"` Text string `json:"text"` + Image string `json:"image,omitempty"` Stream bool `json:"stream"` } @@ -108,12 +111,12 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body")) return } - if strings.TrimSpace(req.Text) == "" { - writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "text is required")) + if strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "" { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "text or image is required")) return } if req.Stream { - s.publishChatStream(r.Context(), req.SessionID, req.Text) + s.publishChatStream(r.Context(), req.SessionID, req.Text, req.Image) writeJSON(w, http.StatusOK, map[string]any{"stream": true}) return } @@ -124,7 +127,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { if req.SessionID == "" { req.SessionID = "web" } - res, err := s.runner.Run(r.Context(), req.SessionID, []model.Message{model.TextMessage(model.RoleUser, req.Text)}) + res, err := s.runner.Run(r.Context(), req.SessionID, userChatMessages(req.Text, req.Image)) if err != nil { s.log.Error("chat run", "err", err, "session", req.SessionID) writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_failed", "chat failed")) @@ -185,6 +188,16 @@ func firstMode(modes []string) string { return modes[0] } +// userChatMessages builds the user turn for a chat / stream request. When image is +// present the turn becomes an image message (with optional text); otherwise a +// plain text message. An empty text with an image yields an image-only turn. +func userChatMessages(text, image string) []model.Message { + if strings.TrimSpace(image) != "" { + return []model.Message{model.ImageMessage(model.RoleUser, text, image)} + } + return []model.Message{model.TextMessage(model.RoleUser, text)} +} + // summaryOfSession maps a session.Session to its wire shape. func summaryOfSession(sess session.Session) SessionSummary { return SessionSummary{ diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index f40e2ff..5f668f5 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -40,6 +40,8 @@ import ( "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" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/automations" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" ) // source is the log provenance tag for every record this package emits. @@ -110,6 +112,26 @@ type TaskOutputStore interface { TaskOutput(ctx context.Context, id string) (task.OutputEvent, error) } +// AutomationStore enumerates the automations and reminders for the admin list. +// *automations.Service is the production implementation. The gateway maps each +// record to its wire shape. +type AutomationStore interface { + List() []automations.AutomationRecord +} + +// UserStore enumerates the accounts and API keys visible to the admin panel. +// The production implementation reads the configured admin account plus the +// auth.APIKey table. The gateway maps each summary to its wire shape. +type UserStore interface { + List() []auth.UserSummary +} + +// BackthinkStore enumerates the persisted background-think records for the +// M9/M10 monitor panel. *backthink.Backthink is the production implementation. +type BackthinkStore interface { + List(ctx context.Context, limit int) ([]backthink.Record, 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 @@ -141,6 +163,13 @@ type Deps struct { // Tasks queries task output for the task_output_query WS route. // Optional (task_output_query returns a clear error when unset). Tasks TaskOutputStore + // Automations lists automations and reminders. Optional. + Automations AutomationStore + // Users lists accounts and API keys. Optional (falls back to the configured + // single admin account). + Users UserStore + // Backthink lists background-think records. Optional. + Backthink BackthinkStore // 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. @@ -165,6 +194,10 @@ type Server struct { iot IoTStore auditor SubagentAuditor taskOutputs TaskOutputStore + automations AutomationStore + users UserStore + backthink BackthinkStore + voice VoiceService streaming bool frontend fs.FS log *slog.Logger @@ -193,6 +226,9 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { iot: deps.IoT, auditor: deps.Scheduler, taskOutputs: deps.Tasks, + automations: deps.Automations, + users: deps.Users, + backthink: deps.Backthink, streaming: deps.Streaming, frontend: deps.Frontend, log: sourcedLog(deps.Log), @@ -229,6 +265,10 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/admin/plugins", adminAuth(http.HandlerFunc(s.adminListPlugins))) mux.Handle("GET /api/v1/admin/channels", adminAuth(http.HandlerFunc(s.adminListChannels))) mux.Handle("GET /api/v1/admin/iot", adminAuth(http.HandlerFunc(s.adminListIoT))) + mux.Handle("GET /api/v1/admin/llm", adminAuth(http.HandlerFunc(s.adminGetLlm))) + mux.Handle("GET /api/v1/admin/automations", adminAuth(http.HandlerFunc(s.adminListAutomations))) + mux.Handle("GET /api/v1/admin/users", adminAuth(http.HandlerFunc(s.adminListUsers))) + mux.Handle("GET /api/v1/admin/backthink", adminAuth(http.HandlerFunc(s.adminListBackthink))) mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy))) var h http.Handler = mux diff --git a/internal/services/gateway/voice.go b/internal/services/gateway/voice.go new file mode 100644 index 0000000..fcf081c --- /dev/null +++ b/internal/services/gateway/voice.go @@ -0,0 +1,163 @@ +package gateway + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice" +) + +// VoiceService runs speech-to-text and text-to-speech for the voice_input WS +// route. *voice.Chain (the degrading provider chain) and *voice.Service (the +// background worker pool) both satisfy it. It is nil-tolerant: the voice route +// replies with a clear error when no service is injected. +type VoiceService interface { + // STT transcribes audio to text. + STT(ctx context.Context, audio voice.Audio) (string, error) + // TTS synthesizes audio from text. + TTS(ctx context.Context, text string) (voice.Audio, error) +} + +// voiceInputRequest is the decoded payload of a voice_input frame. Audio is a +// base64 (or data: URI) audio blob; Format is its encoding ("wav"/"mp3"/"ogg"); +// Text overrides STT when present (a client that already transcribed can skip +// recognition); SessionID defaults to "web". +type voiceInputRequest struct { + Audio string `json:"audio"` + Format string `json:"format"` + Text string `json:"text"` + SessionID string `json:"sessionId"` +} + +// SetVoice injects the voice STT/TTS service used by the voice_input WS route. +// It is the gateway's exposed injection point so the composition root can wire a +// *voice.Chain or *voice.Service without touching Deps. A nil value disables the +// route (it replies with a clear error). +func (s *Server) SetVoice(v VoiceService) { + s.voice = v + s.hub.voice = v +} + +// handleVoiceInput routes one voice_input frame. Without an injected voice +// service it replies with a clear error (never a crash); otherwise it decodes the +// request and runs the STT→chat→TTS turn asynchronously so the read pump is never +// blocked by provider latency. +func (c *wsConn) handleVoiceInput(in wsInbound) { + if c.h.voice == nil { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "voice unavailable"}}) + return + } + var req voiceInputRequest + _ = json.Unmarshal(in.Data, &req) + if strings.TrimSpace(req.Audio) == "" && strings.TrimSpace(req.Text) == "" { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "voice audio or text required"}}) + return + } + go c.h.runVoiceTurn(c, req) +} + +// runVoiceTurn drives one voice turn: STT (or a supplied transcript) → agent chat +// → TTS, then replies to the requesting connection with a voice_output frame +// carrying the base64 audio plus the recognized transcript and assistant reply. +func (h *hub) runVoiceTurn(c *wsConn, req voiceInputRequest) { + ctx := context.Background() + if req.SessionID == "" { + req.SessionID = "web" + } + transcript := strings.TrimSpace(req.Text) + if transcript == "" { + if strings.TrimSpace(req.Audio) == "" { + h.voiceError(c, "voice audio or text required") + return + } + audio, err := decodeVoiceAudio(req.Audio, req.Format) + if err != nil { + h.log.Warn("voice audio decode", "err", err) + h.voiceError(c, "voice audio could not be decoded") + return + } + stt, err := h.voice.STT(ctx, audio) + if err != nil { + h.log.Warn("voice stt", "err", err) + h.voiceError(c, "voice transcription failed") + return + } + transcript = strings.TrimSpace(stt) + } + if transcript == "" { + h.voiceError(c, "no speech recognized") + return + } + if h.runner == nil { + h.voiceError(c, "chat service is not configured") + return + } + res, err := h.runner.Run(ctx, req.SessionID, []model.Message{model.TextMessage(model.RoleUser, transcript)}) + if err != nil { + h.log.Warn("voice chat run", "err", err, "session", req.SessionID) + h.voiceError(c, "voice chat failed") + return + } + reply := res.Text + if strings.TrimSpace(reply) == "" { + reply = res.StopReason + } + audi, err := h.voice.TTS(ctx, reply) + if err != nil { + h.log.Warn("voice tts", "err", err) + h.voiceError(c, "voice synthesis failed") + return + } + c.enqueueJSON(wsEvent{Type: wsVoiceOutput, Data: map[string]any{ + "audio": base64.StdEncoding.EncodeToString(audi.Data), + "format": audi.Format, + "text": transcript, + "reply": reply, + }}) +} + +// voiceError replies with a typed error frame for a failed voice turn. +func (h *hub) voiceError(c *wsConn, message string) { + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": message}}) +} + +// decodeVoiceAudio turns a base64 (or data:) audio payload into voice.Audio. The +// format is taken from req when set, guessed from a data: URI mime otherwise, and +// defaulted to "wav". +func decodeVoiceAudio(src, format string) (voice.Audio, error) { + payload := src + if strings.HasPrefix(src, "data:") { + if comma := strings.IndexByte(src, ','); comma >= 0 { + if format == "" { + format = formatFromMime(src[:comma]) + } + payload = src[comma+1:] + } + } + if strings.TrimSpace(payload) == "" { + return voice.Audio{}, fmt.Errorf("voice: empty audio") + } + data, err := base64.StdEncoding.DecodeString(payload) + if err != nil { + return voice.Audio{}, fmt.Errorf("voice: decode audio: %w", err) + } + if format == "" { + format = "wav" + } + return voice.Audio{Format: format, Data: data}, nil +} + +// formatFromMime guesses an audio format from a data: URI mime segment. +func formatFromMime(meta string) string { + lower := strings.ToLower(meta) + for _, f := range []string{"wav", "mp3", "ogg", "webm", "mpeg"} { + if strings.Contains(lower, f) { + return f + } + } + return "wav" +} diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index 4b8b565..d81e527 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -28,6 +28,7 @@ const ( wsChatSubject = "ws.chat" wsStreamSubject = "ws.stream" wsVoiceInput = "ws.voice_input" + wsVoiceOutput = "voice_output" wsReadLimit = 4096 wsWriteWait = 10 * time.Second wsPongWait = 60 * time.Second @@ -72,6 +73,7 @@ type hub struct { runner Chatter auditor SubagentAuditor taskOutputs TaskOutputStore + voice VoiceService streaming bool log *slog.Logger upgrader websocket.Upgrader @@ -90,6 +92,7 @@ func newHub(s *Server) *hub { runner: s.runner, auditor: s.auditor, taskOutputs: s.taskOutputs, + voice: s.voice, streaming: s.streaming, log: s.log, origins: origins, @@ -213,14 +216,14 @@ func (h *hub) streamChat(ctx context.Context, payload any) { _ = json.Unmarshal(b, &req) } turnID := newTurnID() - if h.runner == nil || strings.TrimSpace(req.Text) == "" { + if h.runner == nil || (strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "") { h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "idle"}}) return } if req.SessionID == "" { req.SessionID = "web" } - res, err := h.runner.Run(ctx, req.SessionID, []model.Message{model.TextMessage(model.RoleUser, req.Text)}) + res, err := h.runner.Run(ctx, req.SessionID, userChatMessages(req.Text, req.Image)) if err != nil { h.log.Error("ws chat run", "err", err, "session", req.SessionID) h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "error", "error": err.Error()}}) @@ -256,6 +259,7 @@ func (h *hub) broadcastStream(payload any) { type chatStreamRequest struct { SessionID string `json:"sessionId"` Text string `json:"text"` + Image string `json:"image,omitempty"` } // handleWS upgrades /ws/chat after authenticating the JWT from the query token or @@ -382,11 +386,11 @@ func (c *wsConn) enqueueJSON(ev wsEvent) { } // publishChatStream publishes a ws.chat streaming trigger (REST stream mode). -func (s *Server) publishChatStream(ctx context.Context, sessionID, text string) { +func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image string) { if s.bus == nil { return } - _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text}) + _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image}) } // wsToken extracts the JWT from the ?token= query or the Authorization header. diff --git a/internal/services/gateway/ws_route.go b/internal/services/gateway/ws_route.go index ea2bfbc..c4843b0 100644 --- a/internal/services/gateway/ws_route.go +++ b/internal/services/gateway/ws_route.go @@ -15,10 +15,11 @@ func (c *wsConn) route(in wsInbound) { 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)) + // Voice routes through the injected STT→chat→TTS pipeline. Without a + // voice service it replies with a clear error rather than dropping the + // frame. The audio reply is delivered as a voice_output event on the + // requesting connection. + c.handleVoiceInput(in) case "subagent_ask": c.handleSubagentAsk(in) case "task_output_query":