diff --git a/internal/runtime/model/openai_stream.go b/internal/runtime/model/openai_stream.go index 2f7fe55..c8ad62c 100644 --- a/internal/runtime/model/openai_stream.go +++ b/internal/runtime/model/openai_stream.go @@ -27,6 +27,22 @@ func (a *openAIToolAcc) idOrName() string { return a.name } +// openAIStreamState is the running aggregation of one streamed +// /chat/completions response: the assembled Response, the bracketing flags of the +// currently open segment, and the per-index tool-call accumulators. It is the +// OpenAI counterpart of anthropic's streamBlock bookkeeping, so both adapters +// share one parse → apply → finalize shape. emit is the nil-tolerant event sink +// (a nil onEvent simply drops every event). +type openAIStreamState struct { + agg Response + text strings.Builder + textStarted bool + thinkingStarted bool + toolCalls map[int]*openAIToolAcc + toolOrder []int + onEvent func(StreamEvent) +} + // scanSSE incrementally decodes stream chunks, accumulates the full body text // with a strings.Builder (never reading the whole body at once) and captures // usage from the final chunk. It splits each delta into its segment and brackets @@ -37,19 +53,7 @@ func (a *openAIToolAcc) idOrName() string { func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes) - agg := Response{} - var text strings.Builder - textStarted := false - thinkingStarted := false - toolCalls := map[int]*openAIToolAcc{} - var toolOrder []int - - endThinking := func() { - if onEvent != nil && thinkingStarted { - thinkingStarted = false - onEvent(StreamEvent{Kind: StreamKindThinkingEnd}) - } - } + st := &openAIStreamState{toolCalls: map[int]*openAIToolAcc{}, onEvent: onEvent} for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) @@ -64,99 +68,143 @@ func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { if err := json.Unmarshal([]byte(data), &chunk); err != nil { return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err) } - if u := chunk.Usage; u != nil { - agg.PromptTokens = u.PromptTokens - agg.CompletionTokens = u.CompletionTokens - agg.TotalTokens = u.TotalTokens - } - if len(chunk.Choices) > 0 && chunk.Choices[0].FinishReason != "" { - agg.StopReason = chunk.Choices[0].FinishReason - } - if len(chunk.Choices) == 0 || chunk.Choices[0].Delta == nil { - continue - } - delta := chunk.Choices[0].Delta - - // Reasoning / thinking segment (OpenAI o-series, DeepSeek, DashScope). - if delta.ReasoningContent != "" { - if onEvent != nil && !thinkingStarted { - thinkingStarted = true - onEvent(StreamEvent{Kind: StreamKindThinkingStart}) - } - if onEvent != nil { - onEvent(StreamEvent{Kind: StreamKindThinkingDelta, Delta: delta.ReasoningContent}) - } - continue - } - - // Streaming tool-call fragments. - if len(delta.ToolCalls) > 0 { - endThinking() - for _, tc := range delta.ToolCalls { - acc := toolCalls[tc.Index] - if acc == nil { - acc = &openAIToolAcc{} - toolCalls[tc.Index] = acc - toolOrder = append(toolOrder, tc.Index) - } - if acc.id == "" { - acc.id = tc.ID - } - if acc.name == "" { - acc.name = tc.Function.Name - } - if onEvent != nil && !acc.started && acc.idOrName() != "" { - acc.started = true - onEvent(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()}) - } - if tc.Function.Arguments != "" { - acc.args.WriteString(tc.Function.Arguments) - if onEvent != nil { - onEvent(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments}) - } - } - } - continue - } - - // Body-text segment. - if delta.Content == "" { - continue - } - endThinking() - text.WriteString(delta.Content) - if onEvent != nil { - if !textStarted { - textStarted = true - onEvent(StreamEvent{Kind: StreamKindTextStart}) - } - onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: delta.Content}) - } + st.applyChunk(chunk) } if err := scanner.Err(); err != nil { return Response{}, fmt.Errorf("model: %w: read stream: %v", ErrUpstream, err) } + return finalizeOpenAIStream(st), nil +} - if onEvent != nil { - if thinkingStarted { - onEvent(StreamEvent{Kind: StreamKindThinkingEnd}) - } - if textStarted { - onEvent(StreamEvent{Kind: StreamKindTextEnd}) - } - // OpenAI has no per-block stop event, so each tool_call block is closed - // when the stream terminates. - for _, idx := range toolOrder { - acc := toolCalls[idx] - if acc == nil || !acc.started { - continue - } - onEvent(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()}) - } +// emit forwards one structured event, tolerating a nil sink. +func (st *openAIStreamState) emit(ev StreamEvent) { + if st.onEvent != nil { + st.onEvent(ev) } +} - for _, idx := range toolOrder { - acc := toolCalls[idx] +// applyChunk folds a single decoded stream chunk into the running aggregation: +// usage and stop reason are recorded even on a choice-less terminal chunk, and +// the segment delta (if any) is routed to its handler. +func (st *openAIStreamState) applyChunk(chunk openAIChatResponse) { + if u := chunk.Usage; u != nil { + st.agg.PromptTokens = u.PromptTokens + st.agg.CompletionTokens = u.CompletionTokens + st.agg.TotalTokens = u.TotalTokens + } + if len(chunk.Choices) == 0 { + return + } + if chunk.Choices[0].FinishReason != "" { + st.agg.StopReason = chunk.Choices[0].FinishReason + } + if chunk.Choices[0].Delta == nil { + return + } + st.applyDelta(chunk.Choices[0].Delta) +} + +// applyDelta routes one delta fragment to its segment handler. The precedence is +// the provider's: a reasoning fragment wins over a tool-call fragment, which wins +// over body text, so a delta never contributes to two segments at once. +func (st *openAIStreamState) applyDelta(delta *openAIDelta) { + switch { + case delta.ReasoningContent != "": + st.applyThinking(delta.ReasoningContent) + case len(delta.ToolCalls) > 0: + st.endThinking() + for _, tc := range delta.ToolCalls { + st.applyToolCall(tc) + } + case delta.Content != "": + st.endThinking() + st.applyText(delta.Content) + } +} + +// applyThinking forwards one reasoning fragment, opening the thinking segment on +// the first fragment only. +func (st *openAIStreamState) applyThinking(fragment string) { + if !st.thinkingStarted { + st.thinkingStarted = true + st.emit(StreamEvent{Kind: StreamKindThinkingStart}) + } + st.emit(StreamEvent{Kind: StreamKindThinkingDelta, Delta: fragment}) +} + +// endThinking closes the open thinking segment (no-op when none is open), so a +// following tool-call or text segment starts a fresh one. +func (st *openAIStreamState) endThinking() { + if !st.thinkingStarted { + return + } + st.thinkingStarted = false + st.emit(StreamEvent{Kind: StreamKindThinkingEnd}) +} + +// applyToolCall folds one streaming tool-call fragment into its per-index +// accumulator, opening the block's start event once its id/name is known and +// forwarding the argument fragments as they arrive. +func (st *openAIStreamState) applyToolCall(tc openAIToolCallDelta) { + acc := st.toolCalls[tc.Index] + if acc == nil { + acc = &openAIToolAcc{} + st.toolCalls[tc.Index] = acc + st.toolOrder = append(st.toolOrder, tc.Index) + } + if acc.id == "" { + acc.id = tc.ID + } + if acc.name == "" { + acc.name = tc.Function.Name + } + if !acc.started && acc.idOrName() != "" { + acc.started = true + st.emit(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()}) + } + if tc.Function.Arguments == "" { + return + } + acc.args.WriteString(tc.Function.Arguments) + st.emit(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments}) +} + +// applyText appends one body-text fragment, opening the text segment on the first +// fragment only. +func (st *openAIStreamState) applyText(fragment string) { + st.text.WriteString(fragment) + if !st.textStarted { + st.textStarted = true + st.emit(StreamEvent{Kind: StreamKindTextStart}) + } + st.emit(StreamEvent{Kind: StreamKindTextDelta, Delta: fragment}) +} + +// closeSegments emits the end event of every segment still open at stream +// termination. OpenAI has no per-block stop event, so each started tool_call +// block is closed here. +func (st *openAIStreamState) closeSegments() { + if st.thinkingStarted { + st.emit(StreamEvent{Kind: StreamKindThinkingEnd}) + } + if st.textStarted { + st.emit(StreamEvent{Kind: StreamKindTextEnd}) + } + for _, idx := range st.toolOrder { + acc := st.toolCalls[idx] + if acc == nil || !acc.started { + continue + } + st.emit(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()}) + } +} + +// assembleToolCalls decodes each accumulated tool-call argument buffer into its +// structured Args map, in provider order. Unparseable fragments degrade to an +// empty map (the call is still surfaced). +func (st *openAIStreamState) assembleToolCalls() { + for _, idx := range st.toolOrder { + acc := st.toolCalls[idx] if acc == nil { continue } @@ -166,9 +214,17 @@ func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { args = map[string]any{} } } - agg.ToolCalls = append(agg.ToolCalls, ToolCall{Name: acc.name, ID: acc.id, Args: args}) + st.agg.ToolCalls = append(st.agg.ToolCalls, ToolCall{Name: acc.name, ID: acc.id, Args: args}) } - agg.Content = text.String() - agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: agg.Content}}} - return agg, nil +} + +// finalizeOpenAIStream closes the open segments and assembles the aggregated +// Response: the concatenated body text (reasoning never leaks into it) plus the +// re-assembled tool calls. +func finalizeOpenAIStream(st *openAIStreamState) Response { + st.closeSegments() + st.assembleToolCalls() + st.agg.Content = st.text.String() + st.agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: st.agg.Content}}} + return st.agg } diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index b96be5c..ef8f209 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -115,14 +115,7 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a // blacklisted/non-whitelisted group or user message never reaches the LLM. The // owner is always allowed (structural, via the resolved actor). actor := session.ResolveActor(msg, b.owner) - if actor.IsOwner() { - // fall through: owner messages are never blacklisted - } else if denied, reason := b.blacklisted(msg, a); denied { - b.log.Warn("inbound message blacklisted, dropped", - "platform", msg.Platform, "channel", msg.ChannelID, "reason", reason) - if b.onRejected != nil { - b.onRejected(msg, reason) - } + if !actor.IsOwner() && b.blacklistDrop(msg, a) { return } @@ -153,17 +146,56 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a } } - // design/16 §4.4 + design/02 §3: an owner private chat is routed INTO the single - // main web session (shared across platforms) and persisted to chat_messages with - // the channel platform, so the Web client mirrors it; group chats and strangers - // keep their isolated sess-xxx context and their own platform log. - sessID := sess.SessionID - isOwnerPrivate := !sess.IsGroup && sess.Owner - platform := "" - if isOwnerPrivate { - sessID = history.MainSessionID - platform = msg.Platform + b.runInboundTurn(ctx, msg, a, text, actor, sess) +} + +// blacklistDrop applies the black-or-white list to a non-owner message, reporting +// whether it must be dropped (design/16 §4.6 §6). A dropped message is audited +// through the OnRejected hook so the decision is visible in the panel. +func (b *Bridge) blacklistDrop(msg adapter.UnifiedMessage, a adapter.PlatformAdapter) bool { + denied, reason := b.blacklisted(msg, a) + if !denied { + return false } + b.log.Warn("inbound message blacklisted, dropped", + "platform", msg.Platform, "channel", msg.ChannelID, "reason", reason) + if b.onRejected != nil { + b.onRejected(msg, reason) + } + return true +} + +// inboundTarget is where one admitted inbound message is persisted and mirrored +// (design/16 §4.4 + design/02 §3): an owner private chat is routed into the single +// main web session with the channel platform stamped so the Web client mirrors it, +// while a restricted group/stranger conversation keeps its isolated sess-xxx +// session and its own platform log. +type inboundTarget struct { + // sessionID is the session the turn runs in. + sessionID string + // private marks an owner private chat mirrored into the main web session. + private bool + // platform is the mirrored chat_messages platform tag (private only). + platform string +} + +// inboundTargetFor resolves the routing target from the structurally resolved +// session and the message's platform. Only a non-group owner conversation is +// mirrored; everything else stays in its isolated session. +func inboundTargetFor(sess session.Session, platform string) inboundTarget { + if !sess.IsGroup && sess.Owner { + return inboundTarget{sessionID: history.MainSessionID, private: true, platform: platform} + } + return inboundTarget{sessionID: sess.SessionID} +} + +// runInboundTurn drives the agent turn for an admitted message: it resolves the +// routing target, builds the model-visible message list, persists the inbound user +// message before the turn (so it is durable even if the channel reply fails), runs +// the loop and hands the result to deliverReply. +func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, + text string, actor adapter.Actor, sess session.Session) { + target := inboundTargetFor(sess, msg.Platform) // The third-party boundary only applies in a restricted (group/stranger) // context: memory is read-only (MemWritable=false), and the sender name is @@ -174,41 +206,48 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a senderName := senderDisplayName(msg, actor) loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName, msg) - // Persist the inbound user message before the turn so it is durable even if the - // channel reply fails. Owner-private messages land in the main web session and - // are broadcast to the Web client; restricted messages land in their isolated - // session (the adapter log page) and are broadcast as a `channel_persisted` event - // (never into the main/`web` session or the web chat). - if isOwnerPrivate { - b.persistToMain(ctx, history.RoleUser, text, "", platform, deviceIDFor(msg), adapterName(a)) + // Owner-private messages land in the main web session and are broadcast to the + // Web client; restricted messages land in their isolated session (the adapter log + // page) and are broadcast as a `channel_persisted` event (never into the + // main/`web` session or the web chat). + if target.private { + b.persistToMain(ctx, history.RoleUser, text, "", target.platform, deviceIDFor(msg), adapterName(a)) } else { b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a)) } - res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), sessID, loopMsgs) + res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), target.sessionID, loopMsgs) if err != nil { b.log.Warn("agent run failed", "platform", msg.Platform, - "session", sessID, "error", err) + "session", target.sessionID, "error", err) return } + b.deliverReply(ctx, msg, a, sess, target, res) +} +// deliverReply persists the assistant reply and sends it back through the adapter +// that received the message. An adapter that does not support outbound text +// (adapter.ErrUnsupported) is a debug-level skip, not a failure: the reply is +// already durable in the history store. +func (b *Bridge) deliverReply(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, + sess session.Session, target inboundTarget, res *loop.Result) { reply := strings.TrimSpace(res.Text) if reply == "" { - b.log.Debug("no reply produced", "platform", msg.Platform, "session", sessID) + b.log.Debug("no reply produced", "platform", msg.Platform, "session", target.sessionID) return } // Persist the assistant reply (owner-private → main session + message_persisted // broadcast; restricted → isolated log + channel_persisted broadcast) then // deliver it back to the channel. - if isOwnerPrivate { - b.persistToMain(ctx, history.RoleAssistant, reply, res.Mood, platform, deviceIDFor(msg), adapterName(a)) + if target.private { + b.persistToMain(ctx, history.RoleAssistant, reply, res.Mood, target.platform, deviceIDFor(msg), adapterName(a)) } else { b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "", adapterName(a)) } if err := b.sendReply(ctx, msg, a, reply); err != nil { if errors.Is(err, adapter.ErrUnsupported) { b.log.Debug("channel does not support outbound reply, skipped", - "platform", msg.Platform, "session", sessID) + "platform", msg.Platform, "session", target.sessionID) return } b.log.Warn("send reply failed", "platform", msg.Platform, "error", err) diff --git a/internal/services/gateway/broadcast.go b/internal/services/gateway/broadcast.go new file mode 100644 index 0000000..05f648f --- /dev/null +++ b/internal/services/gateway/broadcast.go @@ -0,0 +1,78 @@ +package gateway + +import ( + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +// This file owns the gateway's outbound realtime pushes: the persisted-message +// frames shared with the channel bridge plus the per-persona state frames the +// frontend renders (mood badge, capacity awaken). + +// BroadcastPersisted broadcasts a just-persisted chat message as a WS event so +// every browser of the same account appends it without a manual refresh +// (design/16 §4.2). It is shared by the orchestrator (Web turns) and the channel +// bridge (IM messages), so both paths flow through one frame shape. The event type +// is split by session: the main web session and its web sub-sessions broadcast +// `message_persisted` for web-chat realtime sync, while any other session id is an +// isolated channel adapter session (sess-xxx) and broadcasts `channel_persisted` +// for the IM adapter log page — the web chat has no handler for the latter, so it +// is naturally ignored there and never pollutes the conversation. Thinking is +// stripped to stay consistent with the WS history replay (design/16 §6). It is a +// no-op-safe call when the hub has broadcast wired. +func (s *Server) BroadcastPersisted(msg history.Message) { + wire := toHistoryMessage(msg) + wire.Thinking = "" + typ := "channel_persisted" + if msg.SessionID == history.MainSessionID || strings.HasPrefix(msg.SessionID, history.MainSessionID+":") { + typ = "message_persisted" + } + s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) +} + +// wsCapacityAwaken is the WS event type for a capacity state transition +// (design/14 §5). Its data field is the frontend AwakenPayload {fromState,toState,tool}. +const wsCapacityAwaken = "capacity_awaken" + +// BroadcastCapacityAwaken fans a capacity state transition to the Web UI as a +// `capacity_awaken` frame (design/14 §5). sessionID is the conversation that +// transitioned, so the client targets the correct session rather than treating the +// wake event as global. data matches the frontend AwakenPayload {fromState, toState, +// tool}; tool may be empty (no single tool was lit, the client then keeps the prior +// tool badge). +func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool string) { + s.hub.broadcast(wsEvent{ + Type: wsCapacityAwaken, + SessionID: sessionID, + Data: map[string]any{ + "fromState": fromState, + "toState": toState, + "tool": tool, + }, + }) +} + +// wsMoodEvent is the WS event type carrying the persona's mood after a turn +// (design/07 §3 下行 `mood`; design/01 §2 情绪状态机). The frontend's existing +// `mood` handler (ws.ts) reads data.mood and shows the 心情 badge. +const wsMoodEvent = "mood" + +// BroadcastMood fans one emotion change to every Web client as a `mood` frame +// (design/07 §3). It is called only when the mood actually changed — the +// machinery in the composition root is the gate — so the badge does not flicker +// with an unchanged value on every turn. data matches what the frontend `mood` +// handler reads ({mood}); affection/personaId ride along as the honest extra +// detail (affection is the persona's current 好感度, personaId the persona whose +// mood moved) and are ignored by clients that do not render them. The mood is a +// persona-level value, not a per-session one, so no sessionId is attached. +func (s *Server) BroadcastMood(personaID, mood string, affection int) { + s.hub.broadcast(wsEvent{ + Type: wsMoodEvent, + Data: map[string]any{ + "mood": mood, + "personaId": personaID, + "affection": affection, + }, + }) +} diff --git a/internal/services/gateway/contracts.go b/internal/services/gateway/contracts.go new file mode 100644 index 0000000..6f01cb2 --- /dev/null +++ b/internal/services/gateway/contracts.go @@ -0,0 +1,231 @@ +package gateway + +import ( + "context" + "net/http" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" + "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader" + personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" + "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" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/automations" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" +) + +// This file declares the dependency contracts the gateway consumes. Every one is +// a narrow interface satisfied by a concrete backend the composition root injects +// (see Deps in deps.go), so the gateway never reaches for a concrete +// implementation and tests can substitute a stub. + +// Authenticator abstracts the credential + RBAC service. *auth.JWTService is the +// production implementation; it is an interface so tests substitute a mock. +type Authenticator interface { + // Sign issues a token for sub with the given role and lifetime. + Sign(sub string, role auth.Role, ttl time.Duration) (string, error) + // Verify validates a token and returns its claims. + Verify(token string) (*auth.Claims, error) + // Authorize returns an HTTP middleware enforcing at least one required role. + Authorize(required ...auth.Role) func(http.Handler) http.Handler +} + +// CapabilityStore abstracts the global capability registry used to enumerate +// plugins and channels in the admin API. *capability.Registry is the production +// implementation. +type CapabilityStore interface { + // List returns every capability whose Kind matches, or all when kind is + // empty. + List(kind capreg.Kind) []capreg.Capability + // SetPolicy replaces the reachability markers for the named capability. + SetPolicy(name string, p capreg.ToolPolicy) error +} + +// Chatter runs a single agent turn. *loop.Loop is the production implementation. +type Chatter interface { + // Run drives one turn for sessionID and returns the final answer. + Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) +} + +// SessionStore lists known channel sessions. *session.SessionManager is the +// production implementation. +type SessionStore interface { + List(ctx context.Context) ([]session.Session, error) +} + +// PersonaStore lists registered characters (each bound one-to-one to a persona). +// *persona/registry.Registry is the production implementation. The gateway maps +// each Character to its wire summary; the interface deliberately returns the +// domain type so the wire shape stays out of the store contract. +type PersonaStore interface { + List() []personareg.Character + // Persona returns the persona with the given id (to read its visual assets). + // The production registry implements it; the wire summary reads Avatar / + // ChatBackground from it so each role's visuals stay role-scoped. + Persona(id string) (*loader.Persona, bool) +} + +// MemoryStore lists memory entries. *runtime/memory.Store is the production +// implementation. The gateway maps each MemoryItem to its wire summary. +type MemoryStore interface { + List(ctx context.Context) ([]memory.MemoryItem, error) +} + +// HistoryStore is the persistent chat-history store (design/16). *runtime/history.Store +// is the production implementation. When nil the session/message endpoints degrade: list +// sessions falls back to the channel-session view, and the create/messages/delete +// endpoints answer with a clear 501. +type HistoryStore interface { + Append(ctx context.Context, msg history.Message) error + // AppendFull persists msg like Append but returns the complete stored message + // (with the assigned id/timestamp) so the orchestrator can broadcast it for + // cross-end realtime sync (design/16 §4.2). + AppendFull(ctx context.Context, msg history.Message) (history.Message, error) + List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]history.Message, error) + // ListAfter returns up to limit messages strictly newer than after in + // oldest-first order, for forward incremental catch-up after a reconnect + // (the "before" cursor can only walk backwards). platform filters as in List. + ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]history.Message, error) + Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) + ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]history.Message, error) + ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]history.Message, error) + ListAllLog(ctx context.Context, before int64, limit int) ([]history.Message, error) + DeleteSession(ctx context.Context, sessionID string) error + // Search returns session-scoped content matches (newest first) for the + // ?q= message-search path. An empty query yields an empty slice. + Search(ctx context.Context, sessionID, query string, before int64, limit int) ([]history.Message, error) + // UpdateMessage edits a single message's content and returns the updated + // message. An unknown id returns history.ErrMessageNotFound. + UpdateMessage(ctx context.Context, id, content string) (history.Message, error) + // DeleteMessage removes a single message. An unknown id returns + // history.ErrMessageNotFound. + DeleteMessage(ctx context.Context, id string) error + // RenameSession sets a session's display title (main or sub). An unknown id + // returns history.ErrSessionNotFound. + RenameSession(ctx context.Context, sessionID, title string) error + // SetPinned pins/unpins a session (main or sub). An unknown id returns + // history.ErrSessionNotFound. + SetPinned(ctx context.Context, sessionID string, pinned bool) error + NewSubSession(ctx context.Context, ownerID string) (string, error) + ListSessions(ctx context.Context) ([]history.SessionMeta, error) +} + +// IoTStore lists IoT devices. *iot.Manager is the production implementation. +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) +} + +// 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 +} + +// RefreshGrantStore manages opaque refresh tokens (auth.RefreshStore is the +// production implementation). It is the credential behind the refreshToken +// returned by login and the POST /auth/refresh, /auth/logout endpoints. Only +// hashes are persisted; the plaintext token crosses this interface exactly once +// per issue/rotate. Optional: when nil login omits refreshToken and the refresh +// and logout endpoints report a clear not-configured error. +type RefreshGrantStore interface { + // Issue mints a refresh token for userID/role and returns its plaintext + // (visible once) plus its expiry. + Issue(ctx context.Context, userID string, role auth.Role, ttl time.Duration) (string, time.Time, error) + // Rotate validates plaintext and, when opts.Rotate is set, revokes it and + // returns a replacement. It returns auth.ErrInvalidGrant for an unknown, + // revoked, or expired token. + Rotate(ctx context.Context, plaintext string, opts auth.RotateOptions) (auth.RefreshToken, string, error) + // Revoke marks the token matching plaintext revoked, returning whether a + // live token was found. It is idempotent. + Revoke(ctx context.Context, plaintext string) (bool, error) + // RevokeAll deletes every refresh token (global invalidation on secret + // rotation) and returns how many were removed. + RevokeAll(ctx context.Context) (int64, error) + // RevokeByUser deletes every refresh token for userID (logout by access + // token, where the individual grant is unknown) and returns how many were + // removed. + RevokeByUser(ctx context.Context, userID string) (int64, error) +} + +// 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) +} + +// ChannelConfigStore is the runtime-editable channel-adapter configuration +// (channels.json, design/04 §3.1). It is implemented by the composition root's +// channel store: Load returns the effective per-platform accounts (file → env +// fallback), Path reports the channels.json path, and Reload hot-replaces a +// platform's running adapters after a panel save (Bridge.Reload). Load returns +// account configs with real tokens so the owner write path can persist them; the +// admin read path masks them before returning to the client. +type ChannelConfigStore interface { + Path() string + // Load returns the effective per-platform accounts (file → env fallback), + // including the secret tokens needed to persist them. + Load() map[string][]config.ChannelAccount + // Reload rebuilds and hot-replaces the running adapters for platform. + Reload(platform string) error +} + +// OwnerRegistrar registers a channel account id into the owner whitelist +// (configs/users/owner.yaml). It is the structural identity red line: a scan-login +// confirmation stamps the scanner's bound ilink_user_id into the whitelist so the +// channel bridge recognizes it as the owner via actor metadata, never a prompt. +// The composition root implements it (cmd/cyrene's ownerRegistrar) to persist the +// whitelist and hot-update the live owner profile. Registration is best-effort: +// a failure is logged by the caller but does not fail the already-applied +// channels.json persist. +type OwnerRegistrar interface { + RegisterChannelOwner(ctx context.Context, platform, accountID string) error +} + +// RagStore backs the admin knowledge-base RAG module (design/10 §3.12 module 8). +// *runtime/knowledge.Store is the production implementation: it lists/searches +// the project index, imports a document (chunk + index) and deletes one +// (project entry + cascaded chapters). An implementation that cannot write +// reports a clear error (knowledge.ErrWriteUnsupported) rather than a fabricated +// success; the handlers map a missing document to 404. +type RagStore interface { + // ListDocuments enumerates the knowledge-base documents (project-index + // entries). + ListDocuments(ctx context.Context) ([]knowledge.Doc, error) + // Search answers a knowledge-base query (line-level hierarchy + optional + // semantic/vector ranking). + Search(ctx context.Context, query string) ([]knowledge.Hit, error) + // IndexDocument uploads/chunks/embeds a document into the project index. + IndexDocument(ctx context.Context, doc knowledge.Doc) error + // DeleteDocument removes a document (and its chapters) by id, returning + // knowledge.ErrDocumentNotFound when no such document exists. + DeleteDocument(ctx context.Context, id string) error +} diff --git a/internal/services/gateway/deps.go b/internal/services/gateway/deps.go new file mode 100644 index 0000000..77a579c --- /dev/null +++ b/internal/services/gateway/deps.go @@ -0,0 +1,108 @@ +package gateway + +import ( + "io/fs" + "log/slog" + "net/http" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 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 +// *loop.Loop; Sessions/Personas/Memories are satisfied by the concrete session +// manager, persona registry and memory store, and the handlers return [] only +// when a dependency is left nil. +type Deps struct { + // Log is the base logger. Nil uses a discard logger. + Log *slog.Logger + // Auth is the JWT/RBAC service. Required. + Auth Authenticator + // Bus is the in-process event bus the WS hub subscribes to. Optional. + Bus bus.Bus + // Registry enumerates capabilities/plugins/channels. Optional. + Registry CapabilityStore + // Loop runs chat turns. Optional (chat returns an error when unset). + Loop Chatter + // Session lists channel sessions. Optional. + Session SessionStore + // Personas lists personas. Optional. + Personas PersonaStore + // Memories lists memory entries. Optional. + Memories MemoryStore + // History is the persistent chat-history store (design/16). Optional; when nil the + // session/message endpoints degrade (list falls back to channel sessions, others 501). + History HistoryStore + // 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 + // 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 + // Refresh is the opaque refresh-token store. Optional: when nil login omits + // refreshToken and /auth/refresh + /auth/logout answer not-configured. + Refresh RefreshGrantStore + // Backthink lists background-think records. Optional. + Backthink BackthinkStore + // RAG backs the admin knowledge-base RAG module. Optional; list/search + // return empty when unset, and write endpoints report a clear 501. + RAG RagStore + // LLM is the hot-swappable provider stack (models.json, design/04 §3.1). + // Optional: when nil the admin /llm endpoints fall back to the env config so + // pre-existing gateway tests keep working. + LLM *model.RuntimeLLM + // Channels is the runtime-editable channel-adapter config (channels.json). + // Optional: when nil the admin /channels/config endpoints return a clear + // not-configured error, matching how the LLM stack behaves when unset. + Channels ChannelConfigStore + // Plugins is the runtime-installable external tool provider manager + // (plugins.json + live MCP servers). Optional: when nil install answers a + // clear not-implemented error and uninstall falls back to a plain capability + // removal, so pre-existing gateway tests keep working. + Plugins *pluginmgr.Manager + // OwnerRegistrar stamps a scan-login's bound owner id into the owner + // whitelist (configs/users/owner.yaml) and hot-updates the live profile. + // Optional: when nil the weixin confirm flow still persists channels.json but + // skips the whitelist registration (best-effort by design). + OwnerRegistrar OwnerRegistrar + // AuditStore persists the admin audit trail (observability → audit). When nil + // the gateway degrades to an in-memory store so the endpoint still returns + // data; the composition root injects NewSQLiteAuditStore over a.db for real + // persistence. Optional. + AuditStore AuditStore + // DatabasePath is the resolved path of the SQLite database file the + // composition root actually opened (the value derived by resolveSQLitePath, + // which may differ from cfg.DB.SQLitePath when the data dir default applies). + // It backs the admin backup/restore endpoints. Optional; when empty those + // endpoints answer honestly with a clear not-configured error. + DatabasePath string + // DataDir is the resolved data directory (the value resolveDataDir returns, + // which may differ from cfg.DataDir when the OS default applies). It backs + // the image-upload store (/uploads) and its static route. Optional; + // when empty the upload handlers fall back to cfg.DataDir. + DataDir string + // 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. + Frontend fs.FS + // Webhooks are externally-mounted inbound HTTP handlers (endpoint path → + // handler), typically from channel adapters that receive webhook callbacks + // (e.g. webhook). Each is mounted on the ServeMux without gateway auth; the + // handler itself enforces its own signature/authorization (webhook HMAC). + // Optional. + Webhooks map[string]http.Handler +} diff --git a/internal/services/gateway/routes.go b/internal/services/gateway/routes.go new file mode 100644 index 0000000..2cffefd --- /dev/null +++ b/internal/services/gateway/routes.go @@ -0,0 +1,193 @@ +package gateway + +import ( + "net/http" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" +) + +// This file owns the HTTP surface: the route table (REST /api/v1, the realtime +// WebSocket hub, /api/v1/admin/* and externally-mounted webhook callbacks) and +// the outer middleware chain. Route registration is split by audience so the +// RBAC gate guarding each group stays visible: public, authenticated user, +// private history, admin reads and owner-only writes. + +// Handler returns the fully assembled http.Handler: the ServeMux routing +// REST + WS + admin, wrapped in recovery, CORS and request logging. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + + // The RBAC middlewares are built once per role set and shared by every route + // that requires it (owner>admin>viewer, DevDocs/02 §九). + anyAuth := s.auth.Authorize() + ownerAuth := s.auth.Authorize(auth.RoleOwner) + writeAuth := s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin) + + s.registerPublicRoutes(mux) + s.registerUserRoutes(mux, anyAuth, ownerAuth, writeAuth) + s.registerHistoryRoutes(mux, writeAuth) + s.registerAdminRoutes(mux, ownerAuth, writeAuth) + s.mountWebhooks(mux) + + return s.withMiddleware(mux) +} + +// registerPublicRoutes mounts the unauthenticated surface: liveness plus the +// credential endpoints. Refresh/logout are public by design — the opaque refresh +// token is itself the credential, so a client with an expired access token can +// still rotate or revoke it. Both fail closed with 401 invalid_grant on a bad +// token. +func (s *Server) registerPublicRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /api/v1/health", s.handleHealth) + mux.HandleFunc("POST /api/v1/auth/login", s.handleLogin) + mux.HandleFunc("POST /api/v1/auth/refresh", s.handleRefresh) + mux.HandleFunc("POST /api/v1/auth/logout", s.handleLogout) +} + +// registerUserRoutes mounts the authenticated user-facing surface. Chat is a +// write (owner/admin); reads accept any authenticated role so a viewer can still +// list personas/sessions. Audio STT/TTS and image upload are owner/admin-only (the +// owner's private assistant, audited) and are the REST surface native clients use +// to bypass the 4KB WebSocket frame limit. +func (s *Server) registerUserRoutes( + mux *http.ServeMux, + anyAuth func(http.Handler) http.Handler, + ownerAuth func(http.Handler) http.Handler, + writeAuth func(http.Handler) http.Handler, +) { + mux.Handle("POST /api/v1/chat", writeAuth(http.HandlerFunc(s.handleChat))) + mux.Handle("GET /api/v1/sessions", anyAuth(http.HandlerFunc(s.handleListSessions))) + mux.Handle("GET /api/v1/personas", anyAuth(http.HandlerFunc(s.handleListPersonas))) + + mux.Handle("POST /api/v1/audio/transcribe", ownerAuth(http.HandlerFunc(s.handleTranscribe))) + mux.Handle("POST /api/v1/audio/speech", ownerAuth(http.HandlerFunc(s.handleSpeech))) + mux.Handle("POST /api/v1/audio/voice-clone", ownerAuth(http.HandlerFunc(s.handleVoiceClone))) + + // Image upload (B1): a multipart POST that stores a raster image under the + // resolved data dir and returns a same-origin URL usable directly as + // chat.image. Upload is owner/admin + audited; the static route is a + // capability URL (unguessable name) because chat.image / cannot carry a + // bearer token. Both ride the one gateway port (single-binary red line). + mux.Handle("POST /api/v1/upload", writeAuth(http.HandlerFunc(s.handleUpload))) + mux.HandleFunc("GET /api/v1/files/{name}", s.handleUploadFile) +} + +// registerHistoryRoutes mounts the persistent chat-history surface (design/16) +// plus the realtime endpoint. Session/model reads and writes are the owner's +// private history, so owner/admin only (chat uses the same roles); the message +// content (and thinking) is owner/admin-only per design/16 §6, while the session +// metadata in GET /sessions above stays readable by any authenticated role. +// Creating and deleting a sub session mutate persistent state. +func (s *Server) registerHistoryRoutes(mux *http.ServeMux, writeAuth func(http.Handler) http.Handler) { + mux.Handle("GET /api/v1/sessions/{id}/messages", writeAuth(http.HandlerFunc(s.handleListMessages))) + mux.Handle("POST /api/v1/sessions", writeAuth(http.HandlerFunc(s.handleCreateSubSession))) + mux.Handle("DELETE /api/v1/sessions/{id}", writeAuth(http.HandlerFunc(s.handleDeleteSession))) + mux.Handle("PATCH /api/v1/sessions/{id}", writeAuth(http.HandlerFunc(s.handlePatchSession))) + // Single-message edit/delete (Android collaboration batch ③a, D3). A message is + // addressed by its own opaque id; both are owner/admin-gated history writes. + mux.Handle("PATCH /api/v1/messages/{id}", writeAuth(http.HandlerFunc(s.handlePatchMessage))) + mux.Handle("DELETE /api/v1/messages/{id}", writeAuth(http.HandlerFunc(s.handleDeleteMessage))) + + // Realtime. + mux.HandleFunc("/ws/chat", s.handleWS) +} + +// registerAdminRoutes mounts the admin API: reads require owner or admin, the +// audited policy write requires owner, and every other write is reserved for the +// owner (RBAC red line). Each write handler persists to the real backing store or +// answers honestly with a clear "M10 / not-wired" error rather than fabricating a +// success. +func (s *Server) registerAdminRoutes( + mux *http.ServeMux, + ownerAuth func(http.Handler) http.Handler, + writeAuth func(http.Handler) http.Handler, +) { + // Admin reads: owner/admin. + mux.Handle("GET /api/v1/admin/personas", writeAuth(http.HandlerFunc(s.adminListPersonas))) + mux.Handle("GET /api/v1/admin/sessions", writeAuth(http.HandlerFunc(s.adminListSessions))) + mux.Handle("GET /api/v1/admin/memories", writeAuth(http.HandlerFunc(s.adminListMemories))) + mux.Handle("GET /api/v1/admin/plugins", writeAuth(http.HandlerFunc(s.adminListPlugins))) + mux.Handle("GET /api/v1/admin/plugins/installed", writeAuth(http.HandlerFunc(s.adminListInstalledPlugins))) + mux.Handle("GET /api/v1/admin/channels", writeAuth(http.HandlerFunc(s.adminListChannels))) + mux.Handle("GET /api/v1/admin/channels/messages", writeAuth(http.HandlerFunc(s.handleChannelMessages))) + mux.Handle("GET /api/v1/admin/channels/{platform}/messages", writeAuth(http.HandlerFunc(s.handleChannelPlatformLog))) + mux.Handle("GET /api/v1/admin/iot", writeAuth(http.HandlerFunc(s.adminListIoT))) + mux.Handle("GET /api/v1/admin/llm", writeAuth(http.HandlerFunc(s.adminGetLlm))) + mux.Handle("GET /api/v1/admin/automations", writeAuth(http.HandlerFunc(s.adminListAutomations))) + mux.Handle("GET /api/v1/admin/users", writeAuth(http.HandlerFunc(s.adminListUsers))) + mux.Handle("GET /api/v1/admin/backthink", writeAuth(http.HandlerFunc(s.adminListBackthink))) + mux.Handle("GET /api/v1/admin/rag", writeAuth(http.HandlerFunc(s.adminListRag))) + mux.Handle("POST /api/v1/admin/rag/query", writeAuth(http.HandlerFunc(s.adminSearchRag))) + mux.Handle("GET /api/v1/admin/observability/metrics", writeAuth(http.HandlerFunc(s.adminObservabilityMetrics))) + mux.Handle("GET /api/v1/admin/observability/trace", writeAuth(http.HandlerFunc(s.adminObservabilityTrace))) + mux.Handle("GET /api/v1/admin/observability/audit", writeAuth(http.HandlerFunc(s.adminObservabilityAudit))) + // Unified notification feed (design/10 §3 #16): a read-only projection of the + // audit trail + backthink pushes, so the panel's notification page needs no + // new persisted model. + mux.Handle("GET /api/v1/admin/notifications", writeAuth(http.HandlerFunc(s.adminListNotifications))) + mux.Handle("POST /api/v1/admin/plugins/policy", ownerAuth(http.HandlerFunc(s.adminSetPolicy))) + mux.Handle("GET /api/v1/admin/sessions/{id}/export", writeAuth(http.HandlerFunc(s.adminExportSession))) + + // Admin writes: owner-only, audited. + mux.Handle("POST /api/v1/admin/personas", ownerAuth(http.HandlerFunc(s.adminCreatePersona))) + mux.Handle("PUT /api/v1/admin/personas/{id}", ownerAuth(http.HandlerFunc(s.adminUpdatePersona))) + mux.Handle("POST /api/v1/admin/users", ownerAuth(http.HandlerFunc(s.adminCreateUser))) + mux.Handle("POST /api/v1/admin/users/apikey", ownerAuth(http.HandlerFunc(s.adminIssueApiKey))) + mux.Handle("DELETE /api/v1/admin/users/apikey/{id}", ownerAuth(http.HandlerFunc(s.adminRevokeApiKey))) + mux.Handle("POST /api/v1/admin/iot/{id}/control", ownerAuth(http.HandlerFunc(s.adminIoTControl))) + mux.Handle("POST /api/v1/admin/iot/scenarios", ownerAuth(http.HandlerFunc(s.adminIoTScenario))) + mux.Handle("POST /api/v1/admin/iot/rules", ownerAuth(http.HandlerFunc(s.adminIoTRule))) + mux.Handle("POST /api/v1/admin/automations", ownerAuth(http.HandlerFunc(s.adminCreateAutomation))) + mux.Handle("POST /api/v1/admin/automations/{id}/run", ownerAuth(http.HandlerFunc(s.adminRunAutomation))) + mux.Handle("DELETE /api/v1/admin/automations/{id}", ownerAuth(http.HandlerFunc(s.adminDeleteAutomation))) + mux.Handle("POST /api/v1/admin/llm", ownerAuth(http.HandlerFunc(s.adminAddLlmProvider))) + mux.Handle("POST /api/v1/admin/llm/test", ownerAuth(http.HandlerFunc(s.adminTestLlm))) + mux.Handle("POST /api/v1/admin/channels/open", ownerAuth(http.HandlerFunc(s.adminChannelOpen))) + mux.Handle("POST /api/v1/admin/channels/close", ownerAuth(http.HandlerFunc(s.adminChannelClose))) + mux.Handle("POST /api/v1/admin/channels/test", ownerAuth(http.HandlerFunc(s.adminChannelTest))) + mux.Handle("GET /api/v1/admin/channels/{platform}/login-qr", ownerAuth(http.HandlerFunc(s.adminWeixinLoginQR))) + mux.Handle("GET /api/v1/admin/channels/{platform}/login-status", ownerAuth(http.HandlerFunc(s.adminWeixinLoginStatus))) + mux.Handle("GET /api/v1/admin/channels/{platform}/view", ownerAuth(http.HandlerFunc(s.adminChannelPlatformView))) + mux.Handle("GET /api/v1/admin/channels/{platform}/friends", ownerAuth(http.HandlerFunc(s.adminChannelPlatformFriends))) + mux.Handle("POST /api/v1/admin/channels/{platform}/friend/handle", ownerAuth(http.HandlerFunc(s.adminChannelFriendHandle))) + mux.Handle("POST /api/v1/admin/channels/{platform}/message/test", ownerAuth(http.HandlerFunc(s.adminChannelMessageTest))) + mux.Handle("PUT /api/v1/admin/channels/{platform}/config", ownerAuth(http.HandlerFunc(s.adminSaveChannelConfig))) + mux.Handle("DELETE /api/v1/admin/channels/{platform}/config", ownerAuth(http.HandlerFunc(s.adminDeleteChannelAccount))) + mux.Handle("POST /api/v1/admin/plugins/install", ownerAuth(http.HandlerFunc(s.adminPluginInstall))) + mux.Handle("POST /api/v1/admin/plugins/uninstall", ownerAuth(http.HandlerFunc(s.adminPluginUninstall))) + mux.Handle("POST /api/v1/admin/memories", ownerAuth(http.HandlerFunc(s.adminAddMemory))) + mux.Handle("POST /api/v1/admin/memories/search", ownerAuth(http.HandlerFunc(s.adminMemorySearch))) + mux.Handle("POST /api/v1/admin/rag", ownerAuth(http.HandlerFunc(s.adminIndexRag))) + mux.Handle("POST /api/v1/admin/rag/delete", ownerAuth(http.HandlerFunc(s.adminDeleteRag))) + mux.Handle("POST /api/v1/admin/system/revoke-keys", ownerAuth(http.HandlerFunc(s.adminRevokeAllKeys))) + mux.Handle("POST /api/v1/admin/system/rotate-jwt", ownerAuth(http.HandlerFunc(s.adminRotateJWT))) + mux.Handle("POST /api/v1/admin/system/backup", ownerAuth(http.HandlerFunc(s.adminBackup))) + mux.Handle("POST /api/v1/admin/system/restore", ownerAuth(http.HandlerFunc(s.adminRestore))) +} + +// mountWebhooks mounts externally-supplied inbound callbacks (e.g. the webhook +// channel) without gateway auth; each handler enforces its own +// signature/authorization. +func (s *Server) mountWebhooks(mux *http.ServeMux) { + for path, hdl := range s.webhooks { + if path == "" { + continue + } + mux.Handle(path, hdl) + } +} + +// withMiddleware wraps the routed handler in the outer layers, outermost first: +// the upload capability-URL guard, the SPA fallback (when a frontend is +// configured), panic recovery, CORS and request logging. +func (s *Server) withMiddleware(mux http.Handler) http.Handler { + var h http.Handler = mux + h = guardUploadPaths(h) + if s.frontend != nil { + h = s.spa(h) + } + h = s.recover(h) + h = s.cors(h) + h = s.logging(h) + return h +} diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 84aabd7..977b761 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -14,6 +14,11 @@ // owner>admin>viewer RBAC hierarchy (internal/auth). // - Admin writes are audited (who/what/when/result). // - Realtime events follow DevDocs/design/07 §3 (WS protocol). +// +// Layout: contracts.go declares the injected dependency interfaces, deps.go the +// Deps bundle, server.go the Server lifecycle (construct / serve / close), +// routes.go the route table + outer middleware, broadcast.go the outbound +// realtime pushes. package gateway import ( @@ -25,332 +30,19 @@ import ( "net" "net/http" "strconv" - "strings" "time" - "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" - capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" - "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" - "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" - "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader" - personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" - "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" - "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "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" - "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. const source = "gateway" -// Authenticator abstracts the credential + RBAC service. *auth.JWTService is the -// production implementation; it is an interface so tests substitute a mock. -type Authenticator interface { - // Sign issues a token for sub with the given role and lifetime. - Sign(sub string, role auth.Role, ttl time.Duration) (string, error) - // Verify validates a token and returns its claims. - Verify(token string) (*auth.Claims, error) - // Authorize returns an HTTP middleware enforcing at least one required role. - Authorize(required ...auth.Role) func(http.Handler) http.Handler -} - -// CapabilityStore abstracts the global capability registry used to enumerate -// plugins and channels in the admin API. *capability.Registry is the production -// implementation. -type CapabilityStore interface { - // List returns every capability whose Kind matches, or all when kind is - // empty. - List(kind capreg.Kind) []capreg.Capability - // SetPolicy replaces the reachability markers for the named capability. - SetPolicy(name string, p capreg.ToolPolicy) error -} - -// Chatter runs a single agent turn. *loop.Loop is the production implementation. -type Chatter interface { - // Run drives one turn for sessionID and returns the final answer. - Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) -} - -// SessionStore lists known channel sessions. *session.SessionManager is the -// production implementation. -type SessionStore interface { - List(ctx context.Context) ([]session.Session, error) -} - -// PersonaStore lists registered characters (each bound one-to-one to a persona). -// *persona/registry.Registry is the production implementation. The gateway maps -// each Character to its wire summary; the interface deliberately returns the -// domain type so the wire shape stays out of the store contract. -type PersonaStore interface { - List() []personareg.Character - // Persona returns the persona with the given id (to read its visual assets). - // The production registry implements it; the wire summary reads Avatar / - // ChatBackground from it so each role's visuals stay role-scoped. - Persona(id string) (*loader.Persona, bool) -} - -// MemoryStore lists memory entries. *runtime/memory.Store is the production -// implementation. The gateway maps each MemoryItem to its wire summary. -type MemoryStore interface { - List(ctx context.Context) ([]memory.MemoryItem, error) -} - -// HistoryStore is the persistent chat-history store (design/16). *runtime/history.Store -// is the production implementation. When nil the session/message endpoints degrade: list -// sessions falls back to the channel-session view, and the create/messages/delete -// endpoints answer with a clear 501. -type HistoryStore interface { - Append(ctx context.Context, msg history.Message) error - // AppendFull persists msg like Append but returns the complete stored message - // (with the assigned id/timestamp) so the orchestrator can broadcast it for - // cross-end realtime sync (design/16 §4.2). - AppendFull(ctx context.Context, msg history.Message) (history.Message, error) - List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]history.Message, error) - // ListAfter returns up to limit messages strictly newer than after in - // oldest-first order, for forward incremental catch-up after a reconnect - // (the "before" cursor can only walk backwards). platform filters as in List. - ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]history.Message, error) - Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) - ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]history.Message, error) - ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]history.Message, error) - ListAllLog(ctx context.Context, before int64, limit int) ([]history.Message, error) - DeleteSession(ctx context.Context, sessionID string) error - // Search returns session-scoped content matches (newest first) for the - // ?q= message-search path. An empty query yields an empty slice. - Search(ctx context.Context, sessionID, query string, before int64, limit int) ([]history.Message, error) - // UpdateMessage edits a single message's content and returns the updated - // message. An unknown id returns history.ErrMessageNotFound. - UpdateMessage(ctx context.Context, id, content string) (history.Message, error) - // DeleteMessage removes a single message. An unknown id returns - // history.ErrMessageNotFound. - DeleteMessage(ctx context.Context, id string) error - // RenameSession sets a session's display title (main or sub). An unknown id - // returns history.ErrSessionNotFound. - RenameSession(ctx context.Context, sessionID, title string) error - // SetPinned pins/unpins a session (main or sub). An unknown id returns - // history.ErrSessionNotFound. - SetPinned(ctx context.Context, sessionID string, pinned bool) error - NewSubSession(ctx context.Context, ownerID string) (string, error) - ListSessions(ctx context.Context) ([]history.SessionMeta, error) -} - -// IoTStore lists IoT devices. *iot.Manager is the production implementation. -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) -} - -// 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 -} - -// RefreshGrantStore manages opaque refresh tokens (auth.RefreshStore is the -// production implementation). It is the credential behind the refreshToken -// returned by login and the POST /auth/refresh, /auth/logout endpoints. Only -// hashes are persisted; the plaintext token crosses this interface exactly once -// per issue/rotate. Optional: when nil login omits refreshToken and the refresh -// and logout endpoints report a clear not-configured error. -type RefreshGrantStore interface { - // Issue mints a refresh token for userID/role and returns its plaintext - // (visible once) plus its expiry. - Issue(ctx context.Context, userID string, role auth.Role, ttl time.Duration) (string, time.Time, error) - // Rotate validates plaintext and, when opts.Rotate is set, revokes it and - // returns a replacement. It returns auth.ErrInvalidGrant for an unknown, - // revoked, or expired token. - Rotate(ctx context.Context, plaintext string, opts auth.RotateOptions) (auth.RefreshToken, string, error) - // Revoke marks the token matching plaintext revoked, returning whether a - // live token was found. It is idempotent. - Revoke(ctx context.Context, plaintext string) (bool, error) - // RevokeAll deletes every refresh token (global invalidation on secret - // rotation) and returns how many were removed. - RevokeAll(ctx context.Context) (int64, error) - // RevokeByUser deletes every refresh token for userID (logout by access - // token, where the individual grant is unknown) and returns how many were - // removed. - RevokeByUser(ctx context.Context, userID string) (int64, error) -} - -// 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) -} - -// ChannelConfigStore is the runtime-editable channel-adapter configuration -// (channels.json, design/04 §3.1). It is implemented by the composition root's -// channel store: Load returns the effective per-platform accounts (file → env -// fallback), Path reports the channels.json path, and Reload hot-replaces a -// platform's running adapters after a panel save (Bridge.Reload). Load returns -// account configs with real tokens so the owner write path can persist them; the -// admin read path masks them before returning to the client. -type ChannelConfigStore interface { - Path() string - // Load returns the effective per-platform accounts (file → env fallback), - // including the secret tokens needed to persist them. - Load() map[string][]config.ChannelAccount - // Reload rebuilds and hot-replaces the running adapters for platform. - Reload(platform string) error -} - -// OwnerRegistrar registers a channel account id into the owner whitelist -// (configs/users/owner.yaml). It is the structural identity red line: a scan-login -// confirmation stamps the scanner's bound ilink_user_id into the whitelist so the -// channel bridge recognizes it as the owner via actor metadata, never a prompt. -// The composition root implements it (cmd/cyrene's ownerRegistrar) to persist the -// whitelist and hot-update the live owner profile. Registration is best-effort: -// a failure is logged by the caller but does not fail the already-applied -// channels.json persist. -type OwnerRegistrar interface { - RegisterChannelOwner(ctx context.Context, platform, accountID string) error -} - -// RagStore backs the admin knowledge-base RAG module (design/10 §3.12 module 8). -// *runtime/knowledge.Store is the production implementation: it lists/searches -// the project index, imports a document (chunk + index) and deletes one -// (project entry + cascaded chapters). An implementation that cannot write -// reports a clear error (knowledge.ErrWriteUnsupported) rather than a fabricated -// success; the handlers map a missing document to 404. -type RagStore interface { - // ListDocuments enumerates the knowledge-base documents (project-index - // entries). - ListDocuments(ctx context.Context) ([]knowledge.Doc, error) - // Search answers a knowledge-base query (line-level hierarchy + optional - // semantic/vector ranking). - Search(ctx context.Context, query string) ([]knowledge.Hit, error) - // IndexDocument uploads/chunks/embeds a document into the project index. - IndexDocument(ctx context.Context, doc knowledge.Doc) error - // DeleteDocument removes a document (and its chapters) by id, returning - // knowledge.ErrDocumentNotFound when no such document exists. - DeleteDocument(ctx context.Context, id string) 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 -// *loop.Loop; Sessions/Personas/Memories are satisfied by the concrete session -// manager, persona registry and memory store, and the handlers return [] only -// when a dependency is left nil. -type Deps struct { - // Log is the base logger. Nil uses a discard logger. - Log *slog.Logger - // Auth is the JWT/RBAC service. Required. - Auth Authenticator - // Bus is the in-process event bus the WS hub subscribes to. Optional. - Bus bus.Bus - // Registry enumerates capabilities/plugins/channels. Optional. - Registry CapabilityStore - // Loop runs chat turns. Optional (chat returns an error when unset). - Loop Chatter - // Session lists channel sessions. Optional. - Session SessionStore - // Personas lists personas. Optional. - Personas PersonaStore - // Memories lists memory entries. Optional. - Memories MemoryStore - // History is the persistent chat-history store (design/16). Optional; when nil the - // session/message endpoints degrade (list falls back to channel sessions, others 501). - History HistoryStore - // 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 - // 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 - // Refresh is the opaque refresh-token store. Optional: when nil login omits - // refreshToken and /auth/refresh + /auth/logout answer not-configured. - Refresh RefreshGrantStore - // Backthink lists background-think records. Optional. - Backthink BackthinkStore - // RAG backs the admin knowledge-base RAG module. Optional; list/search - // return empty when unset, and write endpoints report a clear 501. - RAG RagStore - // LLM is the hot-swappable provider stack (models.json, design/04 §3.1). - // Optional: when nil the admin /llm endpoints fall back to the env config so - // pre-existing gateway tests keep working. - LLM *model.RuntimeLLM - // Channels is the runtime-editable channel-adapter config (channels.json). - // Optional: when nil the admin /channels/config endpoints return a clear - // not-configured error, matching how the LLM stack behaves when unset. - Channels ChannelConfigStore - // Plugins is the runtime-installable external tool provider manager - // (plugins.json + live MCP servers). Optional: when nil install answers a - // clear not-implemented error and uninstall falls back to a plain capability - // removal, so pre-existing gateway tests keep working. - Plugins *pluginmgr.Manager - // OwnerRegistrar stamps a scan-login's bound owner id into the owner - // whitelist (configs/users/owner.yaml) and hot-updates the live profile. - // Optional: when nil the weixin confirm flow still persists channels.json but - // skips the whitelist registration (best-effort by design). - OwnerRegistrar OwnerRegistrar - // AuditStore persists the admin audit trail (observability → audit). When nil - // the gateway degrades to an in-memory store so the endpoint still returns - // data; the composition root injects NewSQLiteAuditStore over a.db for real - // persistence. Optional. - AuditStore AuditStore - // DatabasePath is the resolved path of the SQLite database file the - // composition root actually opened (the value derived by resolveSQLitePath, - // which may differ from cfg.DB.SQLitePath when the data dir default applies). - // It backs the admin backup/restore endpoints. Optional; when empty those - // endpoints answer honestly with a clear not-configured error. - DatabasePath string - // DataDir is the resolved data directory (the value resolveDataDir returns, - // which may differ from cfg.DataDir when the OS default applies). It backs - // the image-upload store (/uploads) and its static route. Optional; - // when empty the upload handlers fall back to cfg.DataDir. - DataDir string - // 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. - Frontend fs.FS - // Webhooks are externally-mounted inbound HTTP handlers (endpoint path → - // handler), typically from channel adapters that receive webhook callbacks - // (e.g. webhook). Each is mounted on the ServeMux without gateway auth; the - // handler itself enforces its own signature/authorization (webhook HMAC). - // Optional. - Webhooks map[string]http.Handler -} - // 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 { @@ -442,146 +134,6 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { return s, nil } -// Handler returns the fully assembled http.Handler: the ServeMux routing -// REST + WS + admin, wrapped in recovery, CORS and request logging. -func (s *Server) Handler() http.Handler { - mux := http.NewServeMux() - - // Public, no auth. - mux.HandleFunc("GET /api/v1/health", s.handleHealth) - mux.HandleFunc("POST /api/v1/auth/login", s.handleLogin) - // Refresh/logout are public by design: the opaque refresh token is itself the - // credential, so a client with an expired access token can still rotate or - // revoke it. Both fail closed with 401 invalid_grant on a bad token. - mux.HandleFunc("POST /api/v1/auth/refresh", s.handleRefresh) - mux.HandleFunc("POST /api/v1/auth/logout", s.handleLogout) - - // User side, authenticated. Chat is a write (owner/admin); reads accept any - // authenticated role so a viewer can still list personas/sessions. - mux.Handle("POST /api/v1/chat", s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin)(http.HandlerFunc(s.handleChat))) - readAuth := s.auth.Authorize() - mux.Handle("GET /api/v1/sessions", readAuth(http.HandlerFunc(s.handleListSessions))) - mux.Handle("GET /api/v1/personas", readAuth(http.HandlerFunc(s.handleListPersonas))) - - // Audio (voice) STT/TTS. Owner-only (the owner's private voice assistant), - // audited. These are the REST surface for native clients to upload audio and - // fetch synthesized speech, bypassing the 4KB WebSocket frame limit. - mux.Handle("POST /api/v1/audio/transcribe", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.handleTranscribe))) - mux.Handle("POST /api/v1/audio/speech", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.handleSpeech))) - mux.Handle("POST /api/v1/audio/voice-clone", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.handleVoiceClone))) - - // Image upload (B1): a multipart POST that stores a raster image under the - // resolved data dir and returns a same-origin URL usable directly as - // chat.image. Upload is owner/admin + audited; the static route below is a - // capability URL (unguessable name) because chat.image / cannot carry a - // bearer token. Both ride the one gateway port (single-binary red line). - mux.Handle("POST /api/v1/upload", s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin)(http.HandlerFunc(s.handleUpload))) - mux.HandleFunc("GET /api/v1/files/{name}", s.handleUploadFile) - - // Session/model reads and writes are the owner's private history; owner/admin only - // (chat uses the same roles). Session metadata in GET /sessions stays readAuth for - // any authenticated role, but the message content (and thinking) is owner/admin-only - // per design/16 §6. Creating and deleting a sub session mutate persistent state. - historyAuth := s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin) - mux.Handle("GET /api/v1/sessions/{id}/messages", historyAuth(http.HandlerFunc(s.handleListMessages))) - mux.Handle("POST /api/v1/sessions", historyAuth(http.HandlerFunc(s.handleCreateSubSession))) - mux.Handle("DELETE /api/v1/sessions/{id}", historyAuth(http.HandlerFunc(s.handleDeleteSession))) - mux.Handle("PATCH /api/v1/sessions/{id}", historyAuth(http.HandlerFunc(s.handlePatchSession))) - // Single-message edit/delete (Android collaboration batch ③a, D3). A message is - // addressed by its own opaque id; both are owner/admin-gated history writes. - mux.Handle("PATCH /api/v1/messages/{id}", historyAuth(http.HandlerFunc(s.handlePatchMessage))) - mux.Handle("DELETE /api/v1/messages/{id}", historyAuth(http.HandlerFunc(s.handleDeleteMessage))) - - // Realtime. - mux.HandleFunc("/ws/chat", s.handleWS) - - // Admin: reads require owner or admin; the audited policy write requires owner. - adminAuth := s.auth.Authorize(auth.RoleOwner, auth.RoleAdmin) - mux.Handle("GET /api/v1/admin/personas", adminAuth(http.HandlerFunc(s.adminListPersonas))) - mux.Handle("GET /api/v1/admin/sessions", adminAuth(http.HandlerFunc(s.adminListSessions))) - mux.Handle("GET /api/v1/admin/memories", adminAuth(http.HandlerFunc(s.adminListMemories))) - mux.Handle("GET /api/v1/admin/plugins", adminAuth(http.HandlerFunc(s.adminListPlugins))) - mux.Handle("GET /api/v1/admin/plugins/installed", adminAuth(http.HandlerFunc(s.adminListInstalledPlugins))) - mux.Handle("GET /api/v1/admin/channels", adminAuth(http.HandlerFunc(s.adminListChannels))) - mux.Handle("GET /api/v1/admin/channels/messages", adminAuth(http.HandlerFunc(s.handleChannelMessages))) - mux.Handle("GET /api/v1/admin/channels/{platform}/messages", adminAuth(http.HandlerFunc(s.handleChannelPlatformLog))) - 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("GET /api/v1/admin/rag", adminAuth(http.HandlerFunc(s.adminListRag))) - mux.Handle("POST /api/v1/admin/rag/query", adminAuth(http.HandlerFunc(s.adminSearchRag))) - mux.Handle("GET /api/v1/admin/observability/metrics", adminAuth(http.HandlerFunc(s.adminObservabilityMetrics))) - mux.Handle("GET /api/v1/admin/observability/trace", adminAuth(http.HandlerFunc(s.adminObservabilityTrace))) - mux.Handle("GET /api/v1/admin/observability/audit", adminAuth(http.HandlerFunc(s.adminObservabilityAudit))) - // Unified notification feed (design/10 §3 #16): a read-only projection of the - // audit trail + backthink pushes, so the panel's notification page needs no - // new persisted model. - mux.Handle("GET /api/v1/admin/notifications", adminAuth(http.HandlerFunc(s.adminListNotifications))) - mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy))) - - // Admin writes: owner-only, audited. Reads above accept owner/admin; writes - // are reserved for the owner (RBAC red line). Each handler persists to the - // real backing store or answers honestly with a clear "M10 / not-wired" error - // rather than fabricating a success. - ownerAuth := s.auth.Authorize(auth.RoleOwner) - mux.Handle("POST /api/v1/admin/personas", ownerAuth(http.HandlerFunc(s.adminCreatePersona))) - mux.Handle("PUT /api/v1/admin/personas/{id}", ownerAuth(http.HandlerFunc(s.adminUpdatePersona))) - mux.Handle("POST /api/v1/admin/users", ownerAuth(http.HandlerFunc(s.adminCreateUser))) - mux.Handle("POST /api/v1/admin/users/apikey", ownerAuth(http.HandlerFunc(s.adminIssueApiKey))) - mux.Handle("DELETE /api/v1/admin/users/apikey/{id}", ownerAuth(http.HandlerFunc(s.adminRevokeApiKey))) - mux.Handle("POST /api/v1/admin/iot/{id}/control", ownerAuth(http.HandlerFunc(s.adminIoTControl))) - mux.Handle("POST /api/v1/admin/iot/scenarios", ownerAuth(http.HandlerFunc(s.adminIoTScenario))) - mux.Handle("POST /api/v1/admin/iot/rules", ownerAuth(http.HandlerFunc(s.adminIoTRule))) - mux.Handle("POST /api/v1/admin/automations", ownerAuth(http.HandlerFunc(s.adminCreateAutomation))) - mux.Handle("POST /api/v1/admin/automations/{id}/run", ownerAuth(http.HandlerFunc(s.adminRunAutomation))) - mux.Handle("DELETE /api/v1/admin/automations/{id}", ownerAuth(http.HandlerFunc(s.adminDeleteAutomation))) - mux.Handle("POST /api/v1/admin/llm", ownerAuth(http.HandlerFunc(s.adminAddLlmProvider))) - mux.Handle("POST /api/v1/admin/llm/test", ownerAuth(http.HandlerFunc(s.adminTestLlm))) - mux.Handle("POST /api/v1/admin/channels/open", ownerAuth(http.HandlerFunc(s.adminChannelOpen))) - mux.Handle("POST /api/v1/admin/channels/close", ownerAuth(http.HandlerFunc(s.adminChannelClose))) - mux.Handle("POST /api/v1/admin/channels/test", ownerAuth(http.HandlerFunc(s.adminChannelTest))) - mux.Handle("GET /api/v1/admin/channels/{platform}/login-qr", ownerAuth(http.HandlerFunc(s.adminWeixinLoginQR))) - mux.Handle("GET /api/v1/admin/channels/{platform}/login-status", ownerAuth(http.HandlerFunc(s.adminWeixinLoginStatus))) - mux.Handle("GET /api/v1/admin/channels/{platform}/view", ownerAuth(http.HandlerFunc(s.adminChannelPlatformView))) - mux.Handle("GET /api/v1/admin/channels/{platform}/friends", ownerAuth(http.HandlerFunc(s.adminChannelPlatformFriends))) - mux.Handle("POST /api/v1/admin/channels/{platform}/friend/handle", ownerAuth(http.HandlerFunc(s.adminChannelFriendHandle))) - mux.Handle("POST /api/v1/admin/channels/{platform}/message/test", ownerAuth(http.HandlerFunc(s.adminChannelMessageTest))) - mux.Handle("PUT /api/v1/admin/channels/{platform}/config", ownerAuth(http.HandlerFunc(s.adminSaveChannelConfig))) - mux.Handle("DELETE /api/v1/admin/channels/{platform}/config", ownerAuth(http.HandlerFunc(s.adminDeleteChannelAccount))) - mux.Handle("POST /api/v1/admin/plugins/install", ownerAuth(http.HandlerFunc(s.adminPluginInstall))) - mux.Handle("POST /api/v1/admin/plugins/uninstall", ownerAuth(http.HandlerFunc(s.adminPluginUninstall))) - mux.Handle("POST /api/v1/admin/memories", ownerAuth(http.HandlerFunc(s.adminAddMemory))) - mux.Handle("POST /api/v1/admin/memories/search", ownerAuth(http.HandlerFunc(s.adminMemorySearch))) - mux.Handle("POST /api/v1/admin/rag", ownerAuth(http.HandlerFunc(s.adminIndexRag))) - mux.Handle("POST /api/v1/admin/rag/delete", ownerAuth(http.HandlerFunc(s.adminDeleteRag))) - mux.Handle("GET /api/v1/admin/sessions/{id}/export", adminAuth(http.HandlerFunc(s.adminExportSession))) - mux.Handle("POST /api/v1/admin/system/revoke-keys", ownerAuth(http.HandlerFunc(s.adminRevokeAllKeys))) - mux.Handle("POST /api/v1/admin/system/rotate-jwt", ownerAuth(http.HandlerFunc(s.adminRotateJWT))) - mux.Handle("POST /api/v1/admin/system/backup", ownerAuth(http.HandlerFunc(s.adminBackup))) - mux.Handle("POST /api/v1/admin/system/restore", ownerAuth(http.HandlerFunc(s.adminRestore))) - - // External webhook callbacks (e.g. the webhook channel) are mounted without - // gateway auth; each handler enforces its own signature/authorization. - for path, hdl := range s.webhooks { - if path == "" { - continue - } - mux.Handle(path, hdl) - } - - var h http.Handler = mux - h = guardUploadPaths(h) - if s.frontend != nil { - h = s.spa(h) - } - h = s.recover(h) - h = s.cors(h) - h = s.logging(h) - return h -} - // Start launches the gateway in a goroutine, returning a channel that carries the // ListenAndServe result. The server runs until ctx is canceled. func (s *Server) Start(ctx context.Context) (<-chan error, error) { @@ -642,73 +194,6 @@ func newOrchestrator(s *Server) *Orchestrator { }) } -// BroadcastPersisted broadcasts a just-persisted chat message as a WS event so -// every browser of the same account appends it without a manual refresh -// (design/16 §4.2). It is shared by the orchestrator (Web turns) and the channel -// bridge (IM messages), so both paths flow through one frame shape. The event type -// is split by session: the main web session and its web sub-sessions broadcast -// `message_persisted` for web-chat realtime sync, while any other session id is an -// isolated channel adapter session (sess-xxx) and broadcasts `channel_persisted` -// for the IM adapter log page — the web chat has no handler for the latter, so it -// is naturally ignored there and never pollutes the conversation. Thinking is -// stripped to stay consistent with the WS history replay (design/16 §6). It is a -// no-op-safe call when the hub has broadcast wired. -func (s *Server) BroadcastPersisted(msg history.Message) { - wire := toHistoryMessage(msg) - wire.Thinking = "" - typ := "channel_persisted" - if msg.SessionID == history.MainSessionID || strings.HasPrefix(msg.SessionID, history.MainSessionID+":") { - typ = "message_persisted" - } - s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) -} - -// wsCapacityAwaken is the WS event type for a capacity state transition -// (design/14 §5). Its data field is the frontend AwakenPayload {fromState,toState,tool}. -const wsCapacityAwaken = "capacity_awaken" - -// BroadcastCapacityAwaken fans a capacity state transition to the Web UI as a -// `capacity_awaken` frame (design/14 §5). sessionID is the conversation that -// transitioned, so the client targets the correct session rather than treating the -// wake event as global. data matches the frontend AwakenPayload {fromState, toState, -// tool}; tool may be empty (no single tool was lit, the client then keeps the prior -// tool badge). -func (s *Server) BroadcastCapacityAwaken(sessionID, fromState, toState, tool string) { - s.hub.broadcast(wsEvent{ - Type: wsCapacityAwaken, - SessionID: sessionID, - Data: map[string]any{ - "fromState": fromState, - "toState": toState, - "tool": tool, - }, - }) -} - -// wsMoodEvent is the WS event type carrying the persona's mood after a turn -// (design/07 §3 下行 `mood`; design/01 §2 情绪状态机). The frontend's existing -// `mood` handler (ws.ts) reads data.mood and shows the 心情 badge. -const wsMoodEvent = "mood" - -// BroadcastMood fans one emotion change to every Web client as a `mood` frame -// (design/07 §3). It is called only when the mood actually changed — the -// machinery in the composition root is the gate — so the badge does not flicker -// with an unchanged value on every turn. data matches what the frontend `mood` -// handler reads ({mood}); affection/personaId ride along as the honest extra -// detail (affection is the persona's current 好感度, personaId the persona whose -// mood moved) and are ignored by clients that do not render them. The mood is a -// persona-level value, not a per-session one, so no sessionId is attached. -func (s *Server) BroadcastMood(personaID, mood string, affection int) { - s.hub.broadcast(wsEvent{ - Type: wsMoodEvent, - Data: map[string]any{ - "mood": mood, - "personaId": personaID, - "affection": affection, - }, - }) -} - // sourcedLog returns a logger tagged with source "gateway", substituting a // discard logger for nil so Server can always log safely. func sourcedLog(l *slog.Logger) *slog.Logger {