package runtime import ( "context" "crypto/sha256" "encoding/json" "fmt" "log" "path/filepath" "strings" "sync" "time" nruntime "agent-native/runtime" "agent-native/runtime/checkpoint" "agent-native/runtime/filestate" "agent-native/runtime/gates" "agent-native/runtime/hooks" agentllm "agent-native/runtime/llm" "agent-native/runtime/reminders" agentsession "agent-native/runtime/session" agenttools "agent-native/runtime/tools" "agent-system-backend/internal/ctxasm" "agent-system-backend/internal/projectcontext" "agent-system-backend/internal/squadadapter" "agent-system-backend/internal/wfcompile" "agent-system-backend/pkg/extension" core "agent-system-core" ) // directRun 是直连路径在一次 runtime.Run 期间的部件与交付物状态。 // 工具声明仍用真实 schema 发给模型;注册表 schema 保持宽松,避免内核的 // 参数校验把今天能执行的空参数调用挡掉。 type directRun struct { l *LLMRunner rc RunCtx eff core.EffectiveSettings cardflow bool memoryStore memoryStore finalParts []core.Part explicitCard bool deliverables []*core.ArtifactRef usageTotal core.Usage thinkingIdx int textIdx int turnThink strings.Builder turnText strings.Builder thinkStart time.Time thinkEnd time.Time textStart time.Time textEnd time.Time responding bool acting bool queued []agentllm.ToolCall inflight map[string]*toolSlot // 并行执行下按内核事件 CallID 配对 toolMu sync.Mutex // 并行 onEvent 下 inflight/queued/finalParts 互斥 turnBody string evidence []toolEvidence todos core.TodoList plan core.PlanSnapshot closedToolTurn bool checklistTouched bool toolTurnsSince int nudgedPlanless bool factCheckRounds int // E3:事实核查修正轮计数(上限 factCheckMaxRounds) factCheckPrev map[string]bool // E3:上一轮的未核实 token 集合(未变化则不再拦) // E4(R4 升级规则):harness 侧触发的换路提醒——不依赖模型自觉发现 // 自己在原地打转。成败计数在 onEvent 终点统一维护(noteFailureOutcome), // 达阈值注入一次换路指令;检索连续空结果同理(noteEvidenceCall)。 recentOutcomes []bool // E4(R4):最近 N 次工具成败(true=失败),滑动窗口 failWindowNoted bool // E4(R4):窗口判据已提醒(每 run 一次) zeroHitSearches int zeroHitNoted bool // E4(结构化 source_id 账本):证据类工具的成功结果登记为带 ID 的 // 条目,工具输出尾注【src_N|工具】标记——数字溯源的「材料/来源」 // 从此有可指地址。 ledger []evidenceEntry ledgerCount int progress *gates.ProgressReminder checks []core.CheckPart servedModel string switches []string switchMu sync.Mutex // 任务快慢双速(SR-B10 实路径):started 用注入时钟记 run 起点, // nudged/replied/delegated 三个标志驱动「超时仍未反馈则提醒转双速」的一次性触发。 started time.Time nudgedDualSpeedCount int replied bool delegatedBackground bool materialChecked bool // 材料兜底只问一次(runCtx 侧另有幂等,双保险) // 交付物确认链(I2):confirm_choice 发起过选择卡;用户答复后 // pendingDeliverable 非空——回合结束仍未委派则硬性续跑一轮。 confirmIssued bool pendingDeliverable string deliverableNudged bool // 材料整理通知回合(历史含「材料整理」完成标记):结束仍无选择卡且未委派 // 时,硬性要求发起 confirm_choice(弱模型应急,一次性)。 materialNotice bool materialNoticeUnready bool materialNoticeNudged bool materialChoiceOptions []core.HITLChoice } type toolEvidence struct { Name string Status string Summary string VerifyText string // 给验证器的尾部摘要,比界面上的摘要长 } // toolSlot 是一次在途工具调用的过程状态。 // // 并行工具执行下槽位有两个并发访问方:工具 goroutine 回写结果字段, // 内核事件回调(onEvent 的 tool_execution_end)在 toolMu 下读它们装部件。 // 字段一律经 directRun.writeSlot 在 toolMu 下访问,直接写即数据竞争。 // claimed 只在 toolMu 下访问:并行且无 ctx 注入的调用(宿主 UnknownTool // 直调)按认领取槽位,两个 goroutine 不会写同一个槽位、结果不会互相覆盖。 type toolSlot struct { call agentllm.ToolCall start time.Time summary string verifyText string status string truncated bool claimed bool // 已被某次工具执行认领(无注入路径据此避免撞槽) artifact *core.ArtifactRef } type directTool struct { name string run func(context.Context, json.RawMessage) agenttools.Result } func (t directTool) Name() string { return t.name } func (t directTool) Description() string { return t.name } func (t directTool) Parameters() json.RawMessage { return json.RawMessage(`{"type":"object"}`) } func (t directTool) Execute(ctx context.Context, args json.RawMessage) (agenttools.Result, error) { return t.run(ctx, args), nil } func messagesHaveImages(msgs []agentllm.Message) bool { for _, m := range msgs { if len(m.Images) > 0 { return true } } return false } func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model string, msgs []agentllm.Message, decls []agentllm.ToolDecl, streamOpts agentllm.StreamOptions, cardflow bool, eff core.EffectiveSettings, maxTurns, maxTools int, refresh turnRefresh) error { stream = bindProviderRunID(stream, string(rc.Run().ID)) st := &directRun{l: l, rc: rc, eff: eff, cardflow: cardflow, started: l.now(), inflight: map[string]*toolSlot{}} // 记忆端口(N3 WP5):runCtx 暴露 store 时接线(注入由循环按请求完成)。 if ms, ok := rc.(interface{ StoreMemory() memoryStore }); ok && !projectContextEnabled(rc) { st.memoryStore = ms.StoreMemory() } if !projectContextEnabled(rc) { st.loadChecklist() } if msg, ok := lastAddressed(rc.History()); ok && msg.Role == core.RoleSystem { var text string for _, p := range msg.Parts { if p.Type == core.PartText { text += p.Text } } st.pinSubagentArtifacts(text) } st.servedModel = model // Only the latest addressed system notice owns this turn's material choice. // An earlier successful notice must not arm a choice for a later failure. if latest, ok := lastAddressed(rc.History()); ok && latest.Role == core.RoleSystem { for _, p := range latest.Parts { if p.Type != core.PartText || !strings.Contains(p.Text, materialNoticeMarker) { continue } st.materialChoiceOptions = materialOptionsFromNotice(p.Text) st.materialNotice = len(st.materialChoiceOptions) >= 3 st.materialNoticeUnready = !st.materialNotice break } } stream = budgetedStream(stream, rc.Profile().Budget, int(eff.GetNumber("models.chat.maxOutputTokens", 0))) stream = agentllm.WithFallback(stream, ChatFallbacksFor(st.eff, messagesHaveImages(msgs)), st.noteModelSwitch) defer func() { if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok { st.switchMu.Lock() st.usageTotal.Model = st.servedModel st.switchMu.Unlock() sink.AddUsage(st.usageTotal) } }() reg := agenttools.NewRegistry("") for _, d := range decls { name := d.Name reg.Register(directTool{name: name, run: func(ctx context.Context, args json.RawMessage) agenttools.Result { return st.runToolCtx(ctx, name, string(args)) }}) } // 内核工作区工具接入直连路径(裁定 1:尽可能开放工具):行号分页 read、 // 精确 edit/multi_edit(先读后写)、grep/glob/list——沙箱=会话工作区。 // 产品侧同域 fs_* 仍按 runtime.fsTools 门控(隐藏),不与内核名冲突。 if sandbox := conversationSandbox(rc); sandbox != "" { var cpLog *checkpoint.Log var anchor func() string if host, ok := rc.(interface { OpenCheckpoints() (*checkpoint.Log, error) CheckpointAnchor() string }); ok { if lg, err := host.OpenCheckpoints(); err == nil { cpLog = lg a := host.CheckpointAnchor() anchor = func() string { return a } } } ws := agenttools.WorkspaceToolsCheckpointed(sandbox, filestate.New(), cpLog, anchor) for _, wd := range ws.Decls() { name := wd.Name decls = append(decls, agentllm.ToolDecl{Name: wd.Name, Description: wd.Description, Parameters: wd.Parameters}) wsReg := ws reg.Register(directTool{name: name, run: func(ctx context.Context, args json.RawMessage) agenttools.Result { res := wsReg.Execute(ctx, name, args) st.noteKernelResult(ctx, res) // E4:工作区读仅在 inputs/ 材料读时登记账本(isEvidenceRead; // 自产文件的读结果不打标,防编辑循环被尾注污染——实测教训)。 if !res.IsError && (evidenceTools[name] || isEvidenceRead(name, args)) && strings.TrimSpace(res.Content) != "" { if tag, ok := st.tagSource(name, res.Content); ok { res.Content += "\n【" + tag + "】" } } return res }}) } decls = filterPlanDecls(decls, permissionModeOf(rc)) } user := agentllm.Message{Role: agentllm.RoleUser} if len(msgs) > 0 { user = msgs[len(msgs)-1] } var projectErr error var projectManifest *extension.ProjectContextManifest var projectBody string var lastProjectScope string var projectHistoryTainted bool wrapped := func(ctx context.Context, model string, messages []agentllm.Message, _ []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event { fail := func(err error) <-chan agentllm.Event { out := make(chan agentllm.Event, 1) out <- agentllm.Event{Kind: agentllm.KindError, Err: err} close(out) return out } if projectErr != nil { return fail(projectErr) } var manifest *extension.ProjectContextManifest var saveManifest func(context.Context, extension.ProjectContextManifest) error if projectManifest != nil { copy := *projectManifest manifest = © manifest.Model = model scope := projectScopeKey(*manifest) if lastProjectScope != "" && (scope != lastProjectScope || manifest.AuthorizationEpoch == "" || manifest.PublicationEpoch == "") { projectHistoryTainted = true } lastProjectScope = scope if projectHistoryTainted { if len(msgs) == 0 || user.Role != agentllm.RoleUser || projectBody == "" { return fail(fmt.Errorf("trusted project request seed unavailable after scope change")) } // The kernel's conversation still contains assistant/tool text from // the previous authorization epoch. Rebuild this request solely from // the current run's clean seed and freshly resolved project bundle. messages = append([]agentllm.Message(nil), msgs[:len(msgs)-1]...) messages = append(messages, user) messages = reminders.Attach(messages, []string{projectBody}) } if err := checkProjectInputBudget(rc, messages, decls); err != nil { return fail(err) } wire, err := json.Marshal(messages) if err != nil { return fail(err) } manifest.InputSHA256 = fmt.Sprintf("%x", sha256.Sum256(wire)) if ledger, ok := rc.(interface { ContextEvidenceInRequest([]agentllm.Message) []extension.ProjectContextManifestEntry }); ok { manifest.ReadEvidence = ledger.ContextEvidenceInRequest(messages) } attemptID, err := projectcontext.NewAttemptID() if err != nil { return fail(err) } manifest.AttemptID, manifest.Delivery = attemptID, "attempted" if sink, ok := rc.(interface { SaveContextManifest(context.Context, extension.ProjectContextManifest) error }); ok { saveManifest = sink.SaveContextManifest if err := saveManifest(ctx, *manifest); err != nil { return fail(err) } } } if streamOpts.MaxTokens > 0 { opts.MaxTokens = streamOpts.MaxTokens } if streamOpts.MessageShape != "" { opts.MessageShape = streamOpts.MessageShape } in := stream(ctx, model, messages, decls, opts) out := make(chan agentllm.Event, 1) go func() { terminalRecorded := false record := func(status string) error { if manifest == nil || saveManifest == nil { return nil } manifest.Delivery = status receiptCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() return saveManifest(receiptCtx, *manifest) } defer func() { if !terminalRecorded { _ = record("unknown") } close(out) }() for ev := range in { if ev.Kind == agentllm.KindMessageEnd { if ev.Message.Role == "" { ev.Message.Role = agentllm.RoleAssistant } if len(ev.Message.ToolCalls) > 0 && ev.Message.StopReason == "" { ev.Message.StopReason = agentllm.StopToolCalls } } if ev.Kind == agentllm.KindMessageEnd || ev.Kind == agentllm.KindError { status := "unknown" if ev.Kind == agentllm.KindMessageEnd { status = "responded" } if err := record(status); err != nil { terminalRecorded = true select { case out <- agentllm.Event{Kind: agentllm.KindError, Err: err}: case <-ctx.Done(): } return } terminalRecorded = true } select { case out <- ev: case <-ctx.Done(): return } if terminalRecorded { return } } }() return out } window, reserve, buffer := contextWindow(st.eff) // 运行内被清理的旧工具结果完整落盘(占位写路径,模型可 cat 回看)—— // 此前占位只写「见会话记录」,而模型没有读会话记录的工具。 var resultsDir string sandbox := conversationSandbox(rc) if sandbox != "" { resultsDir = filepath.Join(sandbox, ".agent", "results") } // 会话转录为权威上下文(质量 P2-I2):内核 Session 承载跨轮历史—— // 完整工具结果/参数/附件原文随转录持久,不再由界面消息按 160 字摘要 // 重建。首轮(或重播种)用 ctxasm 整段组装做一次性种子;此后每轮只传 // 当前用户消息,系统段与动态环境经 Reminders 刷新。 // 结构分歧(排除消息 / 变体切换 / 重新生成)改变指纹 ⇒ 重播种转录, // 保证转录与消息库一致(regenerate/exclude/候选选择的等价映射)。 // 打开失败降级为无 Session(旧行为:ctxasm 整段组装,不阻断对话)。 var sessStore *agentsession.Store seedRun := true if sandbox != "" && !projectContextEnabled(rc) { transcriptPath := filepath.Join(sandbox, ".agentx", "transcript.jsonl") rev := contextRevision(rc.History()) if opened, oerr := agentsession.Open(transcriptPath); oerr == nil { meta := readSeedMeta(transcriptPath) switch { case opened.Stats().Lines == 0: // 首次播种:本轮回整段组装作为一次性种子并登记指纹。 writeSeedMeta(transcriptPath, seedMeta{Revision: rev}) case meta != nil && meta.Revision != rev: // 结构分歧(排除/变体切换/重新生成):丢弃旧转录,按当前消息库重建。 if rerr := resetTranscript(transcriptPath); rerr == nil { if reopened, rerr2 := agentsession.Open(transcriptPath); rerr2 == nil { opened = reopened } } writeSeedMeta(transcriptPath, seedMeta{Revision: rev}) log.Printf("[runtime] transcript reseed: rev=%s", rev) default: // 正常续跑;或无指纹的既有转录(JournalOnly 时代/崩溃现场): // 采纳为权威,补登记指纹,不破坏崩溃配对修复。 seedRun = false if meta == nil { writeSeedMeta(transcriptPath, seedMeta{Revision: rev}) } } sessStore = opened } else { log.Printf("[runtime] open session transcript: %v", oerr) } } var history []agentllm.Message if sessStore == nil || seedRun { // 无转录权威:整段组装(旧行为);或播种轮:完整历史作一次性种子。 history = append(history, msgs[:len(msgs)-1]...) } // 转录接近窗口时先做结构化压缩(9 段模板),内核请求侧的微清理兜底。 if sessStore != nil && !seedRun { st.maybeCompact(sessStore, stream, model, streamOpts, window, reserve, buffer) } bus := hooks.New() bus.On(func(c hooks.Call) hooks.Outcome { if c.Event != hooks.Stop { return hooks.Outcome{} } d := st.stop(agentllm.Message{Role: agentllm.RoleAssistant, Content: c.LastText}, nruntime.StopInfo{Continuations: c.Continuations, ToolCalls: c.ToolCalls}) if d.Continue { return hooks.Outcome{Decision: hooks.Block, Reason: d.Reason, Replacement: d.Replacement, Notice: d.Notice} } return hooks.Outcome{Decision: hooks.Allow, Notice: d.Notice} }) res, err := nruntime.Run(rc.Context(), nruntime.Config{ Model: model, ToolExecution: toolExecMode(eff), SerialTools: serialTools, History: history, Session: sessStore, JournalOnly: sessStore == nil, // 有转录即为权威;无转录时不涉及 Tools: reg, Stream: wrapped, MaxTurns: maxTurns, Budget: toolBudget(maxTools), RepeatGuard: true, ContinueOnLength: true, MaxLengthContinuations: -1, ContextWindow: window, OutputReserve: reserve, ContextBuffer: buffer, ResultsDir: resultsDir, OnEvent: st.onEvent, Hooks: bus, Reminders: func(turn int) []string { var out []string out = append(out, st.reminders(turn)...) projectErr, projectManifest, projectBody = nil, nil, "" if refresh.ProjectContext != nil { var body string body, projectManifest, projectErr = refresh.ProjectContext() if projectErr == nil && body != "" { projectBody = body out = append(out, body) } } if turn == 0 { if source, ok := rc.(interface{ WorkContext() string }); ok { if note := strings.TrimSpace(source.WorkContext()); note != "" { out = append(out, note) } } } // 快慢双速入口前置(SR-B10 实路径):带附件的任务在第 0 回合就把 // 双速选项摆给模型——模型对开场指令的遵循远好于中途提醒(实测 // MiniMax 无视中途 system-reminder,但对工作方法开场指令照办)。 // 材料整理入口(附件即保证):系统已确定性启动时注入事实说明 // (职责改为快通道反馈+后续交付物),未预启动(文字+小材料)时 // 保留建议式提示,回合结束由 stop 钩子兜底保证。 if turn == 0 && st.eff.GetNumber("runtime.dualSpeedNudgeSeconds", dualSpeedNudgeDefaultSeconds) > 0 && len(rc.Attachments()) > 0 { note := "" if src, ok := rc.(interface{ MaterialNote() string }); ok { note = strings.TrimSpace(src.MaterialNote()) } if note == "" { note = "本任务附带材料,可能属于重任务。若预计要产出正式文档或做深度处理,采用快慢双速:先在当前回复里给用户即时反馈(材料相关性判断、概要或执行计划),再用 delegate_task 的 background=true 把深度工作放到后台执行——收件箱会显示进度卡,完成后系统自动回投结果并汇报。简单任务按常规同步完成即可。" } out = append(out, note) } // 转录权威:系统段在播种时冻结,动态上下文(尾注/环境/新命中模块) // 只在非播种 run 的首个请求注入一次——run 内反复重发没有信息增量。 if sessStore != nil && !seedRun && turn == 0 { out = append(out, refresh.Notes...) if refresh.EnvUpdate != "" { out = append(out, refresh.EnvUpdate) } if sandbox != "" { transcriptPath := filepath.Join(sandbox, ".agentx", "transcript.jsonl") var ids []string for _, b := range refresh.TaskBlocks { ids = append(ids, b.ModuleID) } for _, id := range updateRemindedModules(transcriptPath, ids) { for _, b := range refresh.TaskBlocks { if b.ModuleID == id { out = append(out, "本轮适用的补充指引("+b.ModuleID+"):\n"+b.Body) } } } } } return out }, PermissionAsk: PermissionAskOf(rc), Trace: l.Trace, AgentName: agentNameOf(rc.Profile()), Memory: st.memoryStore, RetryAttempts: retryAttemptsOf(eff), UnknownTool: func(call agentllm.ToolCall) agenttools.Result { return st.runTool(call.Name, call.Arguments) }, }, user) composed := false // composeFinal 默认关闭(裁定 4):转录权威后工作模型自带完整证据,重写层 // 只见 160 字摘要——实测会虚构章节、承诺未做之事。开启或异常兜底时才重写。 composeWanted := st.eff.GetBool("runtime.composeFinal", false) if err == nil && res.Err == nil && len(st.evidence) > 0 && composeWanted { composed = st.composeFinal(GuardProjectStream(rc, currentUserText(rc), stream), model, res.Notice, streamOpts) } if err == nil && res.Err == nil && !composed && res.Notice == "" && strings.TrimSpace(st.turnBody) == "" && len(st.evidence) > 0 { // 兜底:工作轮结束但没有任何给用户的正文(罕见),才用重写层补一段。 composed = st.composeFinal(GuardProjectStream(rc, currentUserText(rc), stream), model, "", streamOpts) } if res.Notice != "" && !composed { st.finalParts = append(st.finalParts, core.Part{Type: core.PartText, Text: res.Notice}) } if err == nil && res.Err == nil && len(st.finalParts) == 0 { st.finalParts = append(st.finalParts, core.Part{Type: core.PartText, Text: "(模型未返回内容)"}) } for _, ref := range st.latestDeliverables() { st.finalParts = append(st.finalParts, core.Part{Type: core.PartArtifactRef, Artifact: ref}) } st.switchMu.Lock() notes := append([]string(nil), st.switches...) st.switchMu.Unlock() if len(notes) > 0 { prefix := make([]core.Part, 0, len(notes)) for _, note := range notes { prefix = append(prefix, core.Part{Type: core.PartText, Text: note, Process: true}) } st.finalParts = append(prefix, st.finalParts...) } _ = rc.SetAssistantParts(st.finalParts) if err != nil { return err } if res.Err != nil { if strings.Contains(res.Err.Error(), "tool calls") { return fmt.Errorf("budget: 工具调用超过上限 %d(profile.budget.maxToolCalls)", maxTools) } if strings.Contains(res.Err.Error(), "max turns") { return fmt.Errorf("budget: 轮次超过上限 %d(profile.budget.maxTurns)", maxTurns) } return res.Err } return nil } // noteKernelResult 镜像 runTool 的结果登记:内核工作区工具不经过产品执行路径, // 摘要/状态/验证文本在这里补记,保证过程面板与终轮验证器看到同样的证据。 func (st *directRun) noteKernelResult(ctx context.Context, res agenttools.Result) { call, ok := nruntime.ToolCallFromContext(ctx) var slot *toolSlot if ok && call.ID != "" { slot = st.slotFor(call.ID) } else { slot = st.slotFor("") } if slot == nil { return } summary := summarize(res.Content, nil) if res.IsError { summary = "error: " + summary } status := statusOf(nil) if res.IsError { status = "error" } verifyText := tailRunes(res.Content, 600) st.writeSlot(slot, func(s *toolSlot) { s.summary = summary s.status = status s.verifyText = verifyText }) } // writeSlot 在 toolMu 下回写在途槽位:工具 goroutine 写、事件回调读的是同一 // 块内存,只有同一把锁才能给出 happens-before。 func (st *directRun) writeSlot(slot *toolSlot, apply func(*toolSlot)) { if slot == nil { return } st.toolMu.Lock() defer st.toolMu.Unlock() apply(slot) } // appendFinalPart 在 toolMu 下追加最终部件:并行工具执行(工件门面卡、卡片 // 工具)与事件回调共享同一片 finalParts。 func (st *directRun) appendFinalPart(parts ...core.Part) { st.toolMu.Lock() defer st.toolMu.Unlock() st.finalParts = append(st.finalParts, parts...) } // markExplicitCard / cardSuppressed 记「模型已自己发过卡片」:写方是卡片工具 // (保序执行,仍与并行工具 goroutine 同时在跑),读方是工件门面卡分支。 func (st *directRun) markExplicitCard() { st.toolMu.Lock() defer st.toolMu.Unlock() st.explicitCard = true } func (st *directRun) cardSuppressed() bool { st.toolMu.Lock() defer st.toolMu.Unlock() return st.explicitCard } // slotFor 按调用 ID 取在途槽位;ID 为空时认领一个尚未被认领的槽位(兼容无 // 注入的调用路径)。返回的指针仅供本次工具执行内写回,不跨事件持有。 func (st *directRun) slotFor(callID string) *toolSlot { st.toolMu.Lock() defer st.toolMu.Unlock() if callID != "" { slot := st.inflight[callID] if slot != nil { slot.claimed = true } return slot } var free *toolSlot for _, slot := range st.inflight { if !slot.claimed { free = slot break } } if free != nil { free.claimed = true } return free } // noteToolDelta 把模型正在编写的工具参数立刻转成会话事件。 // 执行开始前的这段时间原先没有任何工具事件,界面只剩光标。 func (st *directRun) noteModelSwitch(from, to string) { st.switchMu.Lock() defer st.switchMu.Unlock() st.servedModel = to st.switches = append(st.switches, fmt.Sprintf("模型 %s 在输出前失败,已改用 %s", from, to)) } func (st *directRun) noteToolDelta(ev nruntime.Event) { if !st.acting { _ = st.rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseActing}) st.acting = true } _ = st.rc.Emit(core.EvToolCallDelta, core.ToolCallDeltaPayload{ Index: ev.ToolIndex, CallID: ev.CallID, Name: ev.Tool, Delta: ev.Detail, }) } func (st *directRun) onEvent(ev nruntime.Event) { switch ev.Type { case "auto_retry_start": // 重试可观测(N3 WP6):沿用 process part 形态。 st.finalParts = append(st.finalParts, core.Part{Type: core.PartText, Text: "模型在输出前失败,正在重试(" + ev.Text + ")", Process: true}) case "turn_start": st.responding = false st.acting = false st.turnThink.Reset() st.turnText.Reset() st.thinkStart, st.thinkEnd = time.Time{}, time.Time{} st.textStart, st.textEnd = time.Time{}, time.Time{} st.turnBody = "" case "tool_call_delta": st.noteToolDelta(ev) case "text_delta": if !st.responding { _ = st.rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseResponding}) st.responding = true } now := st.l.now() if st.textStart.IsZero() { st.textStart = now } st.textEnd = now st.turnText.WriteString(ev.Text) _ = st.rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: st.textIdx, Text: ev.Text}) case "reasoning_delta": now := st.l.now() if st.thinkStart.IsZero() { st.thinkStart = now } st.thinkEnd = now st.turnThink.WriteString(ev.Text) _ = st.rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Index: st.thinkingIdx, Text: ev.Text}) case "message_end": if ev.PromptTokens > 0 { if src, ok := st.rc.(interface { ContextObservation() *core.ContextObservation }); ok && src.ContextObservation() != nil { ctxasm.NotePromptUsage(src.ContextObservation().EstChars, ev.PromptTokens) } } st.usageTotal.PromptTokens += ev.PromptTokens st.usageTotal.CompletionTokens += ev.CompletionTokens st.usageTotal.TotalTokens += ev.TotalTokens st.usageTotal.ToolCalls += len(ev.Message.ToolCalls) // 计费拆分(N3 WP6):上一请求覆盖、缓存命中累计。 st.usageTotal.LastPromptTokens = ev.PromptTokens st.usageTotal.LastCompletionTokens = ev.CompletionTokens st.usageTotal.CacheReadTokens += ev.CacheReadTokens st.usageTotal.CacheWriteTokens += ev.CacheWriteTokens st.toolMu.Lock() st.queued = append([]agentllm.ToolCall(nil), ev.Message.ToolCalls...) st.toolMu.Unlock() hasCalls := len(ev.Message.ToolCalls) > 0 if st.turnThink.Len() > 0 { st.finalParts = append(st.finalParts, core.Part{Type: core.PartThinking, Text: st.turnThink.String(), Process: true, DurationMs: durationMs(st.thinkStart, st.thinkEnd)}) st.thinkingIdx++ } if hasCalls { st.closedToolTurn = true } if st.turnText.Len() > 0 { body, think := splitThinkBlock(st.turnText.String()) if think != "" { inlineDur := durationMs(st.thinkStart, st.thinkEnd) if inlineDur == 0 { inlineDur = durationMs(st.textStart, st.textEnd) } st.finalParts = append(st.finalParts, core.Part{Type: core.PartThinking, Text: think, Process: true, DurationMs: inlineDur}) st.thinkingIdx++ } if body != "" { st.finalParts = append(st.finalParts, core.Part{Type: core.PartText, Text: body, Process: hasCalls}) st.turnBody = body if !hasCalls { // 面向用户的答复段落已出现:快通道成立,双速提醒不再需要。 st.replied = true } st.textIdx++ } } case "tool_execution_start": if !st.acting { _ = st.rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseActing}) st.acting = true } if len(st.queued) == 0 { return } call, ok := st.takeQueued(ev.CallID) if !ok { return } st.toolMu.Lock() st.inflight[call.ID] = &toolSlot{call: call, start: st.l.now()} st.toolMu.Unlock() _ = st.rc.Emit(core.EvToolCall, core.ToolCallPayload{CallID: call.ID, Name: call.Name, Arguments: call.Arguments}) case "tool_execution_end": st.toolMu.Lock() slot := st.takeInflight(ev.CallID) if slot == nil { st.toolMu.Unlock() return } // 锁内快照结果字段:工具回写(并行工具 goroutine)与本回调分属两个 // goroutine,解锁后再读槽位字段即数据竞争,事件只带快照走。 call, start := slot.call, slot.start status, summary, truncated := slot.status, slot.summary, slot.truncated end := st.l.now() st.finalParts = append(st.finalParts, core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: call.ID, Name: call.Name, Arguments: call.Arguments, Duration: durationMs(start, end)}}, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: call.ID, Status: status, Summary: summary}}, ) if slot.artifact != nil { st.deliverables = append(st.deliverables, slot.artifact) } st.evidence = append(st.evidence, toolEvidence{Name: call.Name, Status: status, Summary: summary, VerifyText: slot.verifyText}) st.toolMu.Unlock() st.noteFailureOutcome(status) _ = st.rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: call.ID, Status: status, Summary: summary, Truncated: truncated, DurationMs: durationMs(start, end)}) } } func (st *directRun) stop(last agentllm.Message, info nruntime.StopInfo) nruntime.StopDecision { // 材料整理兜底(附件即保证):带附件回合结束仍无后台任务且模型也没自行 // 委派(delegatedBackground 尊重模型判断)→ 系统确定性启动,续跑一轮让 // 模型告知用户。ensureMaterial 幂等 + materialChecked 一次性:不续跑循环。 // 实测动机:MiniMax 对入口提示无视、问一句澄清就收场,0 委派 0 卡片, // 用户全程无反馈。 if !st.materialChecked && len(st.rc.Attachments()) > 0 && !st.delegatedBackground { if ens, ok := st.rc.(interface{ EnsureMaterial() string }); ok { st.materialChecked = true if note := strings.TrimSpace(ens.EnsureMaterial()); note != "" { return nruntime.StopDecision{Continue: true, Reason: materialBackstopReason} } } } // 交付物确认链(I2 混合兜底):用户已在选择卡拍板,但回合结束仍未委派 // 后台任务 → 硬性续跑一轮要求启动(一次性)。 if st.pendingDeliverable != "" && !st.delegatedBackground && !st.deliverableNudged { st.deliverableNudged = true return nruntime.StopDecision{Continue: true, Reason: deliverableBackstopReason} } // 材料通知回合应急(弱模型不发 confirm_choice):汇报了却没让用户选 // 交付物、也没委派任何后台 → 硬性要求发起选择卡(一次性)。 if st.materialNotice && !st.confirmIssued && !st.delegatedBackground && !st.materialNoticeNudged { st.materialNoticeNudged = true return nruntime.StopDecision{Continue: true, Reason: materialNoticeBackstopReason} } if block, why := deliverableGateOf(st.turnBody, st.deliverables, st.rc); block { st.demoteLastText() if st.eff.GetBool("runtime.verify", false) { st.addVerifyNote(why) } if info.Continuations < gates.MaxDeliverableGuards { return nruntime.StopDecision{Continue: true, Replacement: discardedClaimNote, Reason: systemCheckReason(deliverableGuardNote)} } return nruntime.StopDecision{Notice: deliverableFallbackNotice} } if !st.eff.GetBool("runtime.verify", false) { return nruntime.StopDecision{} } // 终轮检查只评每个文件的最新版本:修订产生新工件后,被替代的旧版本 // 已不是交付物,继续按旧版评分会让检查永远无法通过(实测死循环根因)。 latest := st.latestDeliverables() // N1 WP5:工件存在/非空/硬事实检查走下沉的 CheckArtifacts(端口在 llm.go 投影; // P5 后 scorer 只拦 lint error 与 lint 未运行)。 if reasons := gates.CheckArtifacts(artifactRefsOf(latest), artifactLookupOf(st.rc), hardFactScorer()); len(reasons) > 0 { return st.continueOrNotice(info, strings.Join(reasons, ";")) } // P5→E3:事实一致性核查——交付物里的百分比/日期/大额数字回溯本轮材料 // (附件解析文本 + 工具证据)。E3 收紧为两轮:第一轮照常给修正机会 // (改值/删除/加「目标/假设」标注——标注后经 MarkedAsEstimate 机械放行); // 第二轮明确这是最后一次拦截并给出标注出路;两轮后不再拦截(记入检查行 // 如实呈现),续写上限(MaxVerifyContinuations)仍兜底防死循环。 if st.eff.GetBool("runtime.factCheck", true) && len(latest) > 0 && info.Continuations < gates.MaxVerifyContinuations { pairs := st.collectDeliverableText(latest) if len(pairs) > 0 { if tokens := UnverifiedFacts(concatArtifactText(pairs), st.materialCorpus(), factExemptions()); len(tokens) > 0 { if notice, cont := st.factCheckGate(tokens); cont { // 与交付物门禁/续写上限分支对齐:检查未通过时,现场那段总结归入 // 过程时间线——修正轮之后的收尾才是最终答复(实测 run_b520bf2e3037 // 未降级导致好总结被前端切分折叠,用户只看到修正流水账)。 st.demoteLastText() st.addVerifyNote("事实核查:" + notice) return nruntime.StopDecision{Continue: true, Reason: systemCheckReason("终轮事实核查发现以下内容在材料中找不到出处,请修正后重新生成/登记交付物,然后重新给出更新后的完整最终答复(先结论、再产出物与要点,可注明本次修正)。\n" + notice)} } } } } verdict, err := st.l.runVerifier(st.rc, st.turnBody, st.evidence, latest) if err != nil { // 验证器没跑成不再是用户可见正文(实测 Notice 会顶掉正式总结)。 // 同源跳过是能力说明,记通过态检查行;真失败记未通过态检查行, // 过程时间线里都可见,最终答复只留模型自己的收尾。 if strings.Contains(err.Error(), "跳过同源验证") { st.addCheck(true, verifierUnavailableNotice) } else { st.addVerifyNote(err.Error()) } return nruntime.StopDecision{} } if !verdict.Passed { reason := strings.TrimSpace(verdict.Reason) if reason == "" { reason = "未说明原因" } if next := strings.TrimSpace(verdict.NextAction); next != "" { reason += "。下一步:" + next } return st.continueOrNotice(info, reason) } st.addCheck(true, strings.TrimSpace(verdict.Reason)) return nruntime.StopDecision{} } // demoteLastText 把最近一段还没标成过程的正文降级。已经是过程的解说不动。 func (st *directRun) demoteLastText() int { for i := len(st.finalParts) - 1; i >= 0; i-- { if st.finalParts[i].Type == core.PartText && !st.finalParts[i].Process { st.finalParts[i].Process = true return i } } return -1 } // latestDeliverables 同名工件只保留最后登记的版本,顺序不变。 // 修订(同名再登记)后旧版本不再是交付物:不进终轮检查、不重复出现在 // 回复的交付清单里。无名工件互不合并。 // 超长工具输出的截断转存(runmgr 的 tool-<调用>.txt,kind=data)不是交付物: // 不进终轮检查、不占交付卡片(实测被评分器当 .txt 交付物误伤,检查连判负)。 func (st *directRun) latestDeliverables() []*core.ArtifactRef { idx := map[string]int{} var out []*core.ArtifactRef for _, ref := range st.deliverables { if ref == nil || strings.TrimSpace(ref.Name) == "" { out = append(out, ref) continue } if isToolDumpArtifact(ref) { continue } if i, ok := idx[ref.Name]; ok { out[i] = ref continue } idx[ref.Name] = len(out) out = append(out, ref) } return out } // isToolDumpArtifact 识别「工具结果超长截断转工件」的落盘(上下文救济,非交付物)。 func isToolDumpArtifact(ref *core.ArtifactRef) bool { return ref.Kind == "data" && strings.HasPrefix(ref.Name, "tool-") } func (st *directRun) addVerifyNote(reason string) { st.addCheck(false, reason) } // addCheck 记下终轮检查并发出 run.check。有尚未降级的回复时,插在它前面, // 这样检查行留在过程区,回复留在过程区外面。 func (st *directRun) addCheck(passed bool, reason string) { part := core.Part{ Type: core.PartCheck, Process: true, Check: &core.CheckPart{Passed: passed, Reason: reason}, } idx := -1 for i := len(st.finalParts) - 1; i >= 0; i-- { if st.finalParts[i].Type == core.PartText && !st.finalParts[i].Process { idx = i break } } if idx >= 0 { st.finalParts = append(st.finalParts, core.Part{}) copy(st.finalParts[idx+1:], st.finalParts[idx:]) st.finalParts[idx] = part } else { st.finalParts = append(st.finalParts, part) } st.checks = append(st.checks, core.CheckPart{Passed: passed, Reason: reason}) _ = st.rc.Emit(core.EvRunCheck, core.RunCheckPayload{Passed: passed, Reason: reason}) } func (st *directRun) continueOrNotice(info nruntime.StopInfo, reason string) nruntime.StopDecision { if info.Continuations < gates.MaxVerifyContinuations { st.demoteLastText() st.addVerifyNote(reason) return nruntime.StopDecision{Continue: true, Reason: systemCheckReason(reason)} } st.addVerifyNote(reason) // 达到续写上限同样收进检查行(未通过态):让用户看见检查确实没过, // 而不是用一句系统提示替换掉模型的最终答复。 st.addCheck(false, verifyCapNotice) return nruntime.StopDecision{} } func (st *directRun) runTool(name, args string) agenttools.Result { return st.runToolCtx(context.Background(), name, args) } // runToolCtx 执行一个工具调用;ctx 携带内核注入的当前调用(并行执行下按 // CallID 定位在途槽位),无注入时回落唯一在途槽位(旧内核/测试直调)。 func (st *directRun) runToolCtx(ctx context.Context, name, args string) agenttools.Result { call, _ := nruntime.ToolCallFromContext(ctx) if call.ID == "" { call = agentllm.ToolCall{Name: name, Arguments: args} } slot := st.slotFor(call.ID) var out string var err error switch call.Name { case toolSquad: if !st.eff.GetBool("runtime.squad", true) { out = "小队编排已被设置禁用(runtime.squad)" break } var a struct{ Goal string } _ = json.Unmarshal([]byte(call.Arguments), &a) dataDir := st.eff.GetString("storage.dataDir", "./data") summary, serr := squadadapter.Orchestrate(st.rc, dataDir, a.Goal) if serr != nil { out = "小队编排失败:" + serr.Error() } else { out = summary } case toolDelegate: if st.materialNoticeUnready { out = "当前材料整理没有可用的续办依据;请先汇报已保存和未完成的范围,等待用户提出下一步目标。" break } var a struct { Agent, Name, Task, Hint string Background bool } _ = json.Unmarshal([]byte(call.Arguments), &a) if a.Background { if sp, ok := st.rc.(interface { SpawnBackground(agentID, name, task, hint string) (core.ID, error) }); ok { // 材料传递(R2):模型自派的后台任务同样带上材料指引——子代理 // 看不到父回合内联的附件全文,需指明检索通道。 task := a.Task if len(st.rc.Attachments()) > 0 { task += delegateMaterialsNote(st.rc.Attachments()) } id, serr := sp.SpawnBackground(subagentProfileID(a.Agent), orDefault(a.Name, "子代理"), task, a.Hint) if serr != nil { out = "子代理失败:" + serr.Error() } else { st.delegatedBackground = true out = fmt.Sprintf("已在后台启动 %s(慢通道):收件箱已建进度卡,完成后结果回投该卡并自动汇报;先把快通道的即时反馈写给用户,不要等待", id) } break } } var res core.AgentResultPayload var serr error if sp, ok := st.rc.(interface { SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) }); ok { res, serr = sp.SpawnAgent(subagentProfileID(a.Agent), orDefault(a.Name, "子代理"), a.Task, a.Hint) } else { res, serr = st.rc.SpawnSubagent(orDefault(a.Name, "子代理"), a.Task, a.Hint) } if serr != nil { out = "子代理失败:" + serr.Error() } else { // F1 修复:回传子代理结论全文(capSubagentResult 已做 32k 截断与 // 超长落盘指针),Summary 只是 160 字摘要——只回摘要会让父代理在 // 摘要的摘要上凭空补细节。 out = orDefault(res.Text, orDefault(res.Summary, "子代理完成")) st.pinSubagentArtifacts(res.Summary + "\n" + res.Text) } case toolConfirmChoice: if st.materialNoticeUnready { out = "当前材料整理没有可用的交付物建议;请先向用户说明已保存的成果和未完成范围,等待用户提出下一步目标。" break } var a struct { Question string `json:"question"` Options []struct { ID, Label, Description, Impact string Recommended bool } `json:"options"` FreeInput *bool `json:"free_input"` TimeoutSec int `json:"timeout_sec"` } if uerr := json.Unmarshal([]byte(call.Arguments), &a); uerr != nil || (!st.materialNotice && (strings.TrimSpace(a.Question) == "" || len(a.Options) < 2)) || (st.materialNotice && len(st.materialChoiceOptions) < 3) { out = "confirm_choice 参数无效:需要 question 与 2-4 个选项(id/label/description/impact)" break } card := core.HITLCard{Kind: core.HITLKindChoice, Question: strings.TrimSpace(a.Question), AllowFreeInput: a.FreeInput == nil || *a.FreeInput} if st.materialNotice { card.Question = "基于整理结果,接下来生成什么?" card.AllowFreeInput = true card.Choices = append([]core.HITLChoice(nil), st.materialChoiceOptions...) } else { for _, o := range a.Options { choice := core.HITLChoice{ID: strings.TrimSpace(o.ID), Label: strings.TrimSpace(o.Label), Description: o.Description, Impact: o.Impact, Recommended: o.Recommended} if choice.ID == "" { choice.ID = choice.Label } if strings.TrimSpace(choice.Impact) == "" { choice.Impact = "按选项说明执行" } card.Choices = append(card.Choices, choice) } if a.TimeoutSec > 0 { secs := a.TimeoutSec card.TimeoutSec = &secs } } st.confirmIssued = true resp, herr := st.rc.AskHITL(card) if herr != nil { out = "选择等待结束(未获得答复):" + herr.Error() + "。请在回复里列出可选项,请用户直接答复。" break } if id := strings.TrimSpace(resp.ChoiceID); id != "" { label := id for _, c := range card.Choices { if c.ID == id { label = c.Label } } st.pendingDeliverable = label out = "用户已选择:" + label + "。立即用 delegate_task 的 background=true 启动对应后台任务(任务描述基于材料整理结果定制,指明输出文件名与必须覆盖的要点),启动后给用户一句确认反馈即可,不要等待完成。" } else if free := strings.TrimSpace(resp.FreeText); free != "" { st.pendingDeliverable = free out = "用户自由输入:" + free + "。若这是正式交付物要求,立即用 delegate_task 的 background=true 启动对应后台任务(描述结合材料整理结果);若是澄清或追问,先如实答复,再按需推进。" } else { out = "用户未给出明确答复。请在回复里列出可选项,请用户直接答复。" } case toolReview: out = st.reviewFindings(call.Arguments) case toolPlan: var p core.PlanSnapshot if uerr := json.Unmarshal([]byte(call.Arguments), &p); uerr == nil { _ = st.rc.UpdatePlan(p) st.plan = p st.checklistTouched = true out = "计划已更新" } else { out = "计划参数错误: " + uerr.Error() } case toolTodos: var t core.TodoList if uerr := json.Unmarshal([]byte(call.Arguments), &t); uerr == nil { _ = st.rc.UpdateTodos(t) st.todos = t st.checklistTouched = true out = "待办已更新" } else { out = "待办参数错误: " + uerr.Error() } case toolRunWorkflow: out = st.runRegisteredWorkflow(call.Arguments) case toolReadProjectEvidence: var a struct { Reference string `json:"reference"` } if e := json.Unmarshal([]byte(call.Arguments), &a); e != nil || strings.TrimSpace(a.Reference) == "" { err = fmt.Errorf("project evidence requires a reference") break } reader, ok := st.rc.(interface { ProjectEvidence(context.Context, string) (extension.ProjectEvidence, error) }) if !ok { err = fmt.Errorf("project evidence unavailable") break } var evidence extension.ProjectEvidence evidence, err = reader.ProjectEvidence(ctx, a.Reference) if err == nil { if evidence.Reference.ID != a.Reference { err = fmt.Errorf("project evidence reference mismatch") } else { out = evidence.Text } } case toolCard: out = st.submitCard(call.Arguments) case toolCardBegin: out = st.beginCard(call.Arguments) case toolCardAppend: out = st.appendCard(call.Arguments) case toolCardCommit: out = st.commitCard() default: res := st.rc.ExecuteTool(core.ToolCallRequest{CallID: call.ID, Name: call.Name, Arguments: json.RawMessage(call.Arguments)}) if res.Status != "ok" { err = fmt.Errorf("%s", res.Output) } else { out = res.Output } // E4:R4 空命中计数 + 证据账本登记(成功且非空的证据类工具输出 // 尾注【src_N|工具】来源标记——数字溯源从此有可指地址)。 st.noteEvidenceCall(call.Name, res.Status, out) if res.Status == "ok" && evidenceTools[call.Name] && strings.TrimSpace(out) != "" { if tag, ok := st.tagSource(call.Name, out); ok { out += "\n【" + tag + "】" } } if res.Artifact != nil && slot != nil { st.writeSlot(slot, func(s *toolSlot) { s.artifact = &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind} }) if st.cardflow && res.Truncated && !st.cardSuppressed() { fc, ferr := st.rc.SubmitCard(artifactFacadeCard(call.Name, *res.Artifact, summarize(out, nil))) if ferr == nil { st.appendFinalPart(core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: fc.ID}}) } } } st.writeSlot(slot, func(s *toolSlot) { s.truncated = res.Truncated }) } summary := summarize(out, nil) status := statusOf(err) if err != nil { summary = "error: " + err.Error() out = summary } verifyText := tailRunes(out, 600) st.writeSlot(slot, func(s *toolSlot) { s.summary = summary s.status = status s.verifyText = verifyText }) if r := []rune(out); len(r) > 16000 { out = string(r[:16000]) + "…(超长截断)" } return agenttools.Result{Content: out, IsError: err != nil} } func (st *directRun) runRegisteredWorkflow(args string) string { var a struct { Workflow string `json:"workflow"` Inputs map[string]string `json:"inputs"` Goal string `json:"goal"` Script string `json:"script"` } if err := json.Unmarshal([]byte(args), &a); err != nil { return "run_workflow 参数不是合法 JSON" } n := 0 for _, s := range []string{a.Workflow, a.Goal, a.Script} { if strings.TrimSpace(s) != "" { n++ } } if n != 1 { return "run_workflow 需要且只能指定 workflow、goal、script 之一" } if strings.TrimSpace(a.Goal) != "" { return st.runPlannedWorkflow(strings.TrimSpace(a.Goal)) } if strings.TrimSpace(a.Script) != "" { return st.runScriptWorkflow(a.Script) } lookup, ok := st.rc.(interface { WorkflowByID(id string) (core.WorkflowDef, bool) }) if !ok { return "当前运行不能查找工作流" } def, ok := lookup.WorkflowByID(a.Workflow) if !ok { return "未注册的工作流 " + a.Workflow } // Q3 WP2:注册工作流经编译图执行(P4 上限在编译入口强制)。 res, err := (CompiledWorkflowRunner{}).runLaunch(st.rc, registeredWorkflowLaunch(st.rc, def, a.Inputs), wfcompile.RunOptions{}) st.pinSubagentArtifacts(res.Summary) if err != nil { if res.Summary != "" { return res.Summary } return err.Error() } return res.Summary } func (st *directRun) runScriptWorkflow(body string) string { // Q3 WP2:模型脚本包成单步工作流走编译图——runtime.workflowScripts // 开关与执行前确认由 wfcompile 宿主处理器持有(P4/legacy 同款:模型脚本 // 只有脚本级确认门,不叠加计划确认)。 res, err := (CompiledWorkflowRunner{}).runLaunch(st.rc, modelScriptLaunch(body), wfcompile.RunOptions{ModelScript: true}) st.pinSubagentArtifacts(res.Summary) if err != nil { if res.Summary != "" { return res.Summary + "\n" + err.Error() } return err.Error() } return res.Summary } func (st *directRun) runPlannedWorkflow(goal string) string { // Q3 WP2:规划分支——规划子代理产出 WorkflowDef,动态上限编译, // 执行前强制计划确认(legacy planWorkflow 同款流)。 res, err := planCompiledWorkflow(st.rc, goal, CompiledWorkflowRunner{}) st.pinSubagentArtifacts(res.Summary) if err != nil { if res.Summary != "" { return res.Summary } return err.Error() } return res.Summary } func (st *directRun) submitCard(args string) string { if !st.cardflow { return "submit_card 未启用(设置 ui.cardflow 关闭);内容直接写在对话回复里" } var a struct{ Title, Summary, Body, Scope string } if err := json.Unmarshal([]byte(args), &a); err != nil || strings.TrimSpace(a.Title) == "" || strings.TrimSpace(a.Body) == "" { return "submit_card 参数错误(title/body 必填): " + errString(err) } body := []core.UINode{{Type: "markdown", Markdown: a.Body}} if strings.TrimSpace(a.Summary) != "" { body = append(body, core.UINode{Type: "kv", KV: [][2]string{{"摘要", a.Summary}}}) } newCard := core.ActivityCard{Kind: core.CardKindResult, Title: a.Title, Status: core.CardDone, Tone: "ok", Body: body} if a.Scope == "project" { newCard.ProjectID = core.ID(st.rc.Run().ProjectID) } card, cerr := st.rc.SubmitCard(newCard) if cerr != nil { return "submit_card 失败: " + cerr.Error() } st.appendFinalPart(core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}}) st.markExplicitCard() where := "对话" if a.Scope == "project" { where = "项目" } return fmt.Sprintf("已作为卡片发出到%s收件箱(%s,card:%s)。对话中保留简短结论即可,不要重复卡片全文。", where, a.Title, card.ID) } func (st *directRun) beginCard(args string) string { if !st.cardflow { return "card_begin 未启用(设置 ui.cardflow 关闭);内容直接写在对话回复里" } var a struct{ Title, Summary, Scope string } if err := json.Unmarshal([]byte(args), &a); err != nil || strings.TrimSpace(a.Title) == "" { return "card_begin 参数错误(title 必填): " + errString(err) } var body []core.UINode if strings.TrimSpace(a.Summary) != "" { body = append(body, core.UINode{Type: "kv", KV: [][2]string{{"摘要", a.Summary}}}) } newCard := core.ActivityCard{Kind: core.CardKindResult, Title: a.Title, Status: core.CardActive, Tone: "ok", Body: body} if a.Scope == "project" { newCard.ProjectID = core.ID(st.rc.Run().ProjectID) } card, cerr := st.rc.BeginStreamCard(newCard) if cerr != nil { return "card_begin 失败: " + cerr.Error() } return fmt.Sprintf("流式卡已开始(%s,card:%s)。用 card_append 逐段追加正文,写完调用 card_commit 收尾。", a.Title, card.ID) } func (st *directRun) appendCard(args string) string { if !st.cardflow { return "card_append 未启用(设置 ui.cardflow 关闭);内容直接写在对话回复里" } var a struct{ Text string } if err := json.Unmarshal([]byte(args), &a); err != nil { return "card_append 参数错误: " + err.Error() } if aerr := st.rc.AppendStreamCard(a.Text); aerr != nil { return "card_append 失败: " + aerr.Error() } return "已追加" } func (st *directRun) commitCard() string { if !st.cardflow { return "card_commit 未启用(设置 ui.cardflow 关闭);内容直接写在对话回复里" } card, cerr := st.rc.CommitStreamCard() if cerr != nil { return "card_commit 失败: " + cerr.Error() } st.appendFinalPart(core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}}) st.markExplicitCard() return fmt.Sprintf("已作为卡片发出(%s,card:%s)。对话中保留简短结论即可,不要重复卡片全文。", card.Title, card.ID) } func (st *directRun) pinSubagentArtifacts(text string) { st.toolMu.Lock() seen := map[string]bool{} for _, ref := range st.deliverables { if ref != nil { seen[ref.ArtifactID] = true } } st.toolMu.Unlock() pinned := make([]*core.ArtifactRef, 0, 4) for _, id := range gates.ArtifactIDPattern.FindAllString(text, -1) { if seen[id] { continue } art, _, ok := st.rc.Artifact(id) if !ok { continue } seen[id] = true pinned = append(pinned, &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}) } if len(pinned) == 0 { return } st.toolMu.Lock() st.deliverables = append(st.deliverables, pinned...) st.toolMu.Unlock() } func errString(err error) string { if err == nil { return "" } return err.Error() } // ---- 并行工具执行的辅助(E1/F5)---- // serialTools 顺序敏感或带会话级副作用的工具:作为内核 SerialTools 传入—— // 并行执行下这些调用按模型声明顺序逐个执行(互斥且保序;只互斥不保序会让 // 流式卡 card_begin→append→commit 乱序,实测踩过)。卡片流四步有先后依赖, // 计划/待办是整表更新,工作流与小队编排派生 run,执行原语与文件编辑共享会话 // 工作区。读检索类与 delegate_task/review_findings 并行。 var serialTools = map[string]bool{ toolPlan: true, toolTodos: true, toolCard: true, toolCardBegin: true, toolCardAppend: true, toolCardCommit: true, toolRunWorkflow: true, toolSquad: true, toolConfirmChoice: true, "run_shell": true, "run_python": true, "register_artifact": true, "render_document": true, "edit": true, "multi_edit": true, } // toolExecMode 同回合多工具调用的执行模式:runtime.parallelTools(缺省开) // 为 parallel(顺序敏感工具经 Config.SerialTools 由内核保序),关闭时空串 // 回落内核缺省 sequential。 func toolExecMode(eff core.EffectiveSettings) string { if eff.GetBool("runtime.parallelTools", true) { return nruntime.ExecParallel } return "" } // takeQueued 按内核事件 CallID 取出队列中对应调用(并行下 start 事件乱序到达); // CallID 为空(旧内核/未知来源)时按原先进先出。调用方须持有 toolMu。 func (st *directRun) takeQueued(callID string) (agentllm.ToolCall, bool) { if callID != "" { for i, c := range st.queued { if c.ID == callID { st.queued = append(st.queued[:i], st.queued[i+1:]...) return c, true } } return agentllm.ToolCall{}, false } if len(st.queued) == 0 { return agentllm.ToolCall{}, false } c := st.queued[0] st.queued = st.queued[1:] return c, true } // takeInflight 按事件 CallID 取出并删除在途槽位;CallID 为空且恰有一个在途 // 槽位时取之(兼容旧内核)。调用方须持有 toolMu。 func (st *directRun) takeInflight(callID string) *toolSlot { if callID != "" { slot := st.inflight[callID] delete(st.inflight, callID) return slot } if len(st.inflight) == 1 { for _, slot := range st.inflight { clear(st.inflight) return slot } } return nil }