feat: 唤醒式架构批次——capacity/L4/会话编排/管理面板写CRUD/前端集成（77-81）
【capacity 唤醒式核心(77)】internal/runtime/capacity：(Decider L1启发式+L2上下文+人格覆盖→三态静息/唤醒/全能→允许工具集+capacity_awaken事件)。loop 外围允许集过滤(不硬阻断=能力≠边界红线)。config 加 Capacity(Enabled/MinConfidence/ToolSets)。wire buildCapacity 接入 loop。 【L4 记忆驱动意图(80)】recall.go：从记忆召回用户习惯(搞一下xx=办事)→唤醒L1/L2漏掉/升置信；群聊陌生人不读owner习惯(隔离红线)；RecordIntent闭环(判定写回记忆越用越准)；默认保守静息。 【会话消息编排(78)】gateway/session_orchestrator.go：每会话(模式/pending有界5/tail上下文尾巴20)；Submit非阻塞(idle立即/busy入队ack/满拒绝)；完成回调按序续接(tail+新消息合并增量注入非全量重放)；Interrupt中断；REST 202/409+WS session_busy/pending_ack事件；不重造任务引擎复用loop。 【前端集成(79)】ws.ts 加 capacity_awaken/session_busy/pending_ack 解析；store 加 awakenState/sessionBusy；TopBar/ChatShell 加唤醒态徽标+忙态提示(不阻塞输入框)；状态组件 status.tsx；Miuix风格。 【管理面板写CRUD(81)】20个写端点(POST/PUT/DELETE /admin/*，owner+审计落库)：personas(真持久化YAML)/users apikey(签发掩码+撤销)/iot控制(真下发)/automations(建/触发/删)/llm(test真调用+临时)/channels(connect/close/身份测试actor白名单)/plugins(卸载)/memories(search真检索)/sessions(export)。无存储如实501不伪造。拆4文件满足≤500红线。前端9模块接真写。 【wire收口】buildCapacity接入loop+OnCapacity事件；server.go newOrchestrator接hub。 实机验证：iot ac.living control 200 真实改变 on:true；observability/metrics 200、health ok；全仓 build/vet/test 47包绿+前端tsc/vite。
This commit is contained in:
Aska committed
1 parent
e0a543cc37
commit
6dbb4a2e18
51 files changed
+5395
-287
No files matched your search
@@ -74,14 +74,15 @@ type hub struct {
|
||||
runner Chatter
|
||||
auditor SubagentAuditor
|
||||
taskOutputs TaskOutputStore
|
||||
voice VoiceService
|
||||
streaming bool
|
||||
log *slog.Logger
|
||||
upgrader websocket.Upgrader
|
||||
origins map[string]bool
|
||||
allowAll bool
|
||||
cancel context.CancelFunc
|
||||
streamSeq atomic.Uint64
|
||||
voice VoiceService
|
||||
streaming bool
|
||||
orchestrator *Orchestrator
|
||||
log *slog.Logger
|
||||
upgrader websocket.Upgrader
|
||||
origins map[string]bool
|
||||
allowAll bool
|
||||
cancel context.CancelFunc
|
||||
streamSeq atomic.Uint64
|
||||
}
|
||||
|
||||
// newHub builds a hub bound to the server's dependencies.
|
||||
@@ -124,11 +125,15 @@ func (h *hub) start() {
|
||||
}
|
||||
}
|
||||
|
||||
// close cancels the bus subscription and shuts every live connection down.
|
||||
// close cancels the bus subscription, interrupts every orchestrator session (so
|
||||
// no background turn outlives the hub), and shuts every live connection down.
|
||||
func (h *hub) close() {
|
||||
if h.cancel != nil {
|
||||
h.cancel()
|
||||
}
|
||||
if h.orchestrator != nil {
|
||||
h.orchestrator.InterruptAll()
|
||||
}
|
||||
h.mu.Lock()
|
||||
conns := make([]*wsConn, 0, len(h.conns))
|
||||
for c := range h.conns {
|
||||
@@ -205,20 +210,19 @@ func (h *hub) relay(ctx context.Context, e bus.Event) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// streamChat runs one agent turn for a chat trigger and broadcasts the turn
|
||||
// terminus (stream_end) with a fresh turnId; on the non-streaming fallback path
|
||||
// it also emits a single stream_chunk. When the loop is streaming, the per-event
|
||||
// deltas arrive via the OnStream→bus→relay path as structured message.*/
|
||||
// thinking.*/tool_call.* events. It runs on its own goroutine so a slow model
|
||||
// streamChat is the ws.chat trigger entry point. It decodes the request, and
|
||||
// routes the turn through the session orchestrator when one is wired (the
|
||||
// non-blocking busy/pending/resume path of design/15); otherwise it falls back to
|
||||
// the legacy direct run so a hub built without an orchestrator keeps the exact
|
||||
// stream_chunk/stream_end contract. It runs on its own goroutine so a slow model
|
||||
// call does not stall the bus subscription worker.
|
||||
func (h *hub) streamChat(ctx context.Context, payload any) {
|
||||
var req chatStreamRequest
|
||||
if b, err := json.Marshal(payload); err == nil {
|
||||
_ = json.Unmarshal(b, &req)
|
||||
}
|
||||
turnID := newTurnID()
|
||||
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"}})
|
||||
h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "idle"}})
|
||||
return
|
||||
}
|
||||
if req.SessionID == "" {
|
||||
@@ -226,21 +230,55 @@ func (h *hub) streamChat(ctx context.Context, payload any) {
|
||||
}
|
||||
// The Web main session is the owner's private conversation (JWT = owner), so
|
||||
// it binds the full-access env: memory read+write, unchanged behavior.
|
||||
res, err := h.runner.Run(loop.WithSessionEnv(ctx, loop.OwnerEnv()), 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()}})
|
||||
msgs := userChatMessages(req.Text, req.Image)
|
||||
if h.orchestrator != nil {
|
||||
_ = h.orchestrator.Submit(context.Background(), req.SessionID, msgs, loop.OwnerEnv())
|
||||
return
|
||||
}
|
||||
// When the loop is streaming (OnStream wired), the structured deltas already
|
||||
// fan out as message.*/thinking.*/tool_call.* events through the
|
||||
// OnStream→bus→relay path; a second whole-text chunk would duplicate the
|
||||
// tail. The non-streaming path (or a stub runner) still emits one chunk so
|
||||
// the stream contract holds in tests.
|
||||
_, _ = h.runTurn(ctx, req.SessionID, msgs, loop.OwnerEnv())
|
||||
}
|
||||
|
||||
// runTurn executes one agent turn and broadcasts its output stream. It is the
|
||||
// unit both the orchestrator (via its injected TurnRunner) and the legacy
|
||||
// fallback path share, so a turn always emits the same stream_chunk/stream_end
|
||||
// events regardless of who drives it. When the loop is streaming (OnStream wired)
|
||||
// the structured deltas already fan out as message.*/thinking.*/tool_call.*
|
||||
// events through the OnStream→bus→relay path, so only the non-streaming path
|
||||
// emits a single whole-text chunk to hold the stream contract.
|
||||
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
|
||||
turnID := newTurnID()
|
||||
if h.runner == nil || len(msgs) == 0 {
|
||||
h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "idle"}})
|
||||
return &loop.Result{StopReason: "idle"}, nil
|
||||
}
|
||||
res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), sessionID, msgs)
|
||||
if err != nil {
|
||||
h.log.Error("chat run", "err", err, "session", sessionID)
|
||||
h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "error", "error": err.Error()}})
|
||||
return nil, err
|
||||
}
|
||||
if !h.streaming {
|
||||
h.broadcast(wsEvent{Type: "stream_chunk", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"delta": res.Text}})
|
||||
}
|
||||
h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}})
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// handleInterrupt routes a client "interrupt" frame (design/15 §3) to the
|
||||
// orchestrator, canceling the in-flight turn and clearing the pending queue for
|
||||
// the requested session. It is a no-op when no orchestrator is wired.
|
||||
func (h *hub) handleInterrupt(payload json.RawMessage) {
|
||||
if h.orchestrator == nil {
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
SessionID string `json:"sessionId"`
|
||||
}
|
||||
_ = json.Unmarshal(payload, &req)
|
||||
if req.SessionID == "" {
|
||||
return
|
||||
}
|
||||
h.orchestrator.Interrupt(req.SessionID)
|
||||
}
|
||||
|
||||
// broadcastStream maps a streamed model.StreamEvent to its downstream WS event
|
||||
|
||||
Reference in new issue
Block a user