Guard active runs with current account state

Why: Disabling an account must stop in-flight work as well as new HTTP requests.

Evidence: SELF receipts /tmp/iam-u3-u2-runtime-regression.json and /tmp/iam-u3-u2-app-compile.json.

Impact: Run start, tools, HITL, streaming output, child runs and final settlement recheck the account independently of project authorization.

Refs: IAM U3 r1 stages 4-5; Dev feature/iam-u3-account-directory-recovery
This commit is contained in:
2026-10-14 03:20:41 +00:00
parent 6c0289faf7
commit d2efea2845
3 changed files with 317 additions and 27 deletions
+12 -1
View File
@@ -7,6 +7,7 @@ import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log"
"net"
@@ -450,6 +451,17 @@ func Run(opt Options) error {
subRunner := &runtime.SubLLMRunner{Prompts: promptSvc, Trace: traceFn}
mockRunner := &runtime.MockRunner{}
mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{})
authOn := resolved.AuthEnabled()
mgr.SetAccountEffectiveAuthorization(func(_ context.Context, owner string) (bool, error) {
account, err := st.GetAccountByOwner(owner)
if errors.Is(err, store.ErrNotFound) && !authOn && owner == store.OwnerDefault {
return true, nil // legacy implicit local singleton has no users row
}
if err != nil {
return false, err
}
return account.Enabled, nil
})
instanceID := opt.InstanceID
if instanceID == "" {
host, err := os.Hostname()
@@ -693,7 +705,6 @@ func Run(opt Options) error {
return int(setSvc.Effective(store.OwnerDefault, "").GetNumber("runtime.rateLimitPerMin", 0))
})
// 认证策略(auto):PG 档默认开启(多用户面不能裸奔);嵌入式档默认关闭(本地单用户)。
authOn := resolved.AuthEnabled()
srv.Auth = server.NewAuthenticator(st, authOn)
if authOn {
log.Printf("auth: API key required (create with: agent-system apikey create --owner <name>)")
+145
View File
@@ -0,0 +1,145 @@
package runmgr
import (
"context"
"encoding/json"
"errors"
"testing"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// A missing domain provider must not make a registered account guard optional.
func TestAccountGuardBlocksRunToolAndQueuedHITL(t *testing.T) {
c, st := newExecEnv(t)
if err := st.CreateAccount("alice", "Alice", "", false); err != nil {
t.Fatal(err)
}
c.mgr.SetAccountEffectiveAuthorization(func(_ context.Context, owner string) (bool, error) {
a, err := st.GetAccountByOwner(owner)
return err == nil && a.Enabled, err
})
c.run.ID = store.NewID("run")
c.run.Status = core.RunWaitingHITL
if err := st.CreateRun(&c.run); err != nil {
t.Fatal(err)
}
h := &runHandle{hitlCh: make(chan hitlAnswer, 1)}
c.mgr.active = map[string]*runHandle{c.run.ID: h}
card := core.HITLCard{ID: "disabled-card", RunID: c.run.ID, ConversationID: c.run.ConversationID}
if err := st.SaveHITLCard("alice", &card); err != nil {
t.Fatal(err)
}
if _, err := st.DB().Exec(`UPDATE users SET enabled=FALSE WHERE owner='alice'`); err != nil {
t.Fatal(err)
}
conv, err := st.GetConversation("alice", c.run.ConversationID)
if err != nil {
t.Fatal(err)
}
if _, err := c.mgr.StartConversationRun(conv, core.Message{}, nil, nil); !errors.Is(err, store.ErrNotFound) {
t.Fatalf("disabled account started run without domain provider: %v", err)
}
res := c.ExecuteTool(core.ToolCallRequest{CallID: "disabled-tool", Name: "write_artifact", Arguments: json.RawMessage(`{"name":"denied.txt","content":"x"}`)})
if res.Status != "error" {
t.Fatalf("disabled account executed tool: %+v", res)
}
arts, err := st.ListArtifacts("alice", c.run.ProjectID)
if err != nil || len(arts) != 0 {
t.Fatalf("disabled account wrote artifact: %+v %v", arts, err)
}
if _, _, err := c.mgr.ResolveHITL("alice", card.ID, core.HITLResponse{ChoiceID: "yes"}); !errors.Is(err, store.ErrNotFound) {
t.Fatalf("disabled account resolved HITL: %v", err)
}
persisted, err := st.GetHITLCard("alice", card.ID)
if err != nil || persisted.Resolved || len(h.hitlCh) != 0 {
t.Fatalf("disabled HITL changed state: %+v channel=%d err=%v", persisted, len(h.hitlCh), err)
}
payload, _ := json.Marshal(HITLCommandPayload{Owner: "alice", CardID: card.ID, Kind: "yes"})
c.mgr.applyHITLCommand(ownedCommand{CommandID: "queued-disabled", Payload: string(payload)})
if len(h.hitlCh) != 0 {
t.Fatal("queued approval delivered to disabled run")
}
}
func TestAccountGuardStoreReadFailureBlocksOutput(t *testing.T) {
c, st := newExecEnv(t)
runCtx, cancel := context.WithCancel(context.Background())
defer cancel()
c.ctx = runCtx
c.mgr.active = map[string]*runHandle{"active": {owner: "alice", cancel: cancel}}
readErr := errors.New("account store unavailable")
c.mgr.SetAccountEffectiveAuthorization(func(context.Context, string) (bool, error) { return false, readErr })
if err := c.Emit(core.EvToolOutput, core.ToolOutputPayload{Text: "secret"}); !errors.Is(err, readErr) {
t.Fatalf("store failure emitted output: %v", err)
}
if runCtx.Err() == nil {
t.Fatal("account-state read failure left active local run executing")
}
if err := c.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "secret"}}); !errors.Is(err, readErr) {
t.Fatalf("store failure retained assistant output: %v", err)
}
evs, err := st.EventsAfter(c.run.ConversationID, 0, 100)
if err != nil || len(evs) != 0 {
t.Fatalf("store failure persisted output events: %+v %v", evs, err)
}
}
func TestAccountGuardBlocksRestoredNoticeOutput(t *testing.T) {
c, st := newExecEnv(t)
if err := st.CreateAccount("alice", "Alice", "", false); err != nil {
t.Fatal(err)
}
c.mgr.SetAccountEffectiveAuthorization(func(_ context.Context, owner string) (bool, error) {
a, err := st.GetAccountByOwner(owner)
return err == nil && a.Enabled, err
})
if _, err := st.DB().Exec(`UPDATE users SET enabled=FALSE WHERE owner='alice'`); err != nil {
t.Fatal(err)
}
if err := c.mgr.startNoticeRun(c.run.ConversationID, []bgNotice{{owner: "alice", name: "restored", payload: core.AgentResultPayload{Summary: "must not deliver"}}}); !errors.Is(err, store.ErrNotFound) {
t.Fatalf("restored notice delivered to disabled account: %v", err)
}
msgs, err := st.ListMessages(c.run.ConversationID)
if err != nil || len(msgs) != 0 {
t.Fatalf("restored notice persisted output: %+v %v", msgs, err)
}
}
func TestAccountGuardDiscardsComposedOutputAfterDisable(t *testing.T) {
c, st := newExecEnv(t)
if err := st.CreateAccount("alice", "Alice", "", false); err != nil {
t.Fatal(err)
}
c.mgr.SetAccountEffectiveAuthorization(func(_ context.Context, owner string) (bool, error) {
a, err := st.GetAccountByOwner(owner)
return err == nil && a.Enabled, err
})
c.run.ID = store.NewID("run")
c.run.Status = core.RunRunning
if err := st.CreateRun(&c.run); err != nil {
t.Fatal(err)
}
c.h = &runHandle{assistant: []core.Part{{Type: core.PartText, Text: "must not persist"}}}
if _, err := st.DB().Exec(`UPDATE users SET enabled=FALSE WHERE owner='alice'`); err != nil {
t.Fatal(err)
}
c.mgr.finishRun(c, core.RunCompleted, core.Usage{}, nil)
got, err := st.GetRunByID(c.run.ID)
if err != nil || got.Status == core.RunCompleted {
t.Fatalf("disabled run completed: %+v %v", got, err)
}
msgs, err := st.ListMessages(c.run.ConversationID)
if err != nil {
t.Fatal(err)
}
for _, msg := range msgs {
for _, part := range msg.Parts {
if part.Text == "must not persist" {
t.Fatal("previously composed assistant output persisted after disable")
}
}
}
}
+160 -26
View File
@@ -57,10 +57,11 @@ type Manager struct {
retriever core.Retriever // 可选混合检索(nil = store 全文)
// sharedKnowledgeAllow EXT-13:回合内检索的共享授权谓词(请求 owner × 知识所属
// projectID);nil = 未注册(现状:仅本人两路,不触碰共享通道)。
sharedKnowledgeAllow func(owner string, projectID core.ID) bool
knowledgeResourceAllow func(owner string, doc core.KnowledgeDoc) bool
knowledgeProjectScope func(context.Context, string, core.ID) (core.ID, error)
projectWorkAuthorization func(context.Context, string, core.ID) (bool, error)
sharedKnowledgeAllow func(owner string, projectID core.ID) bool
knowledgeResourceAllow func(owner string, doc core.KnowledgeDoc) bool
knowledgeProjectScope func(context.Context, string, core.ID) (core.ID, error)
projectWorkAuthorization func(context.Context, string, core.ID) (bool, error)
accountEffectiveAuthorization func(context.Context, string) (bool, error)
// commands EXT-14:命令即工具的投影与分发源(nil = 未接线,命令不进工具清单)。
commands CommandSource
indexer core.DocumentIndexer // 可选索引入口(nil = store 整文档单块)
@@ -132,6 +133,7 @@ type bgNotice struct {
var ErrConversationBusy = fmt.Errorf("runmgr: conversation already has an in-flight run")
type runHandle struct {
owner string
cancel context.CancelFunc
hitlCh chan hitlAnswer
assistant []core.Part
@@ -217,6 +219,42 @@ func (m *Manager) SetKnowledgeResourceAllow(fn func(string, core.KnowledgeDoc) b
func (m *Manager) SetProjectWorkAuthorization(fn func(context.Context, string, core.ID) (bool, error)) {
m.projectWorkAuthorization = fn
}
// SetAccountEffectiveAuthorization binds the current account-state guard. It is
// independent of the optional domain project provider.
func (m *Manager) SetAccountEffectiveAuthorization(fn func(context.Context, string) (bool, error)) {
m.accountEffectiveAuthorization = fn
}
func (m *Manager) authorizeAccountEffective(ctx context.Context, owner string) error {
if m.accountEffectiveAuthorization == nil {
return nil
}
ok, err := m.accountEffectiveAuthorization(ctx, owner)
if err != nil {
m.cancelActiveOwnerRuns(owner)
return err
}
if !ok {
m.cancelActiveOwnerRuns(owner)
return store.ErrNotFound
}
return nil
}
func (m *Manager) cancelActiveOwnerRuns(owner string) {
m.mu.Lock()
var cancels []context.CancelFunc
for _, h := range m.active {
if h.owner == owner && h.cancel != nil {
cancels = append(cancels, h.cancel)
}
}
m.mu.Unlock()
for _, cancel := range cancels {
cancel()
}
}
func (m *Manager) authorizeProjectWork(ctx context.Context, owner string, id core.ID) error {
if m.projectWorkAuthorization == nil {
return nil
@@ -231,6 +269,13 @@ func (m *Manager) authorizeProjectWork(ctx context.Context, owner string, id cor
return nil
}
func (m *Manager) authorizeRunWork(ctx context.Context, owner string, id core.ID) error {
if err := m.authorizeAccountEffective(ctx, owner); err != nil {
return err
}
return m.authorizeProjectWork(ctx, owner, id)
}
// SetKnowledgeProjectScope binds host project filters to the provider's trusted
// canonical domain metadata without widening the current conversation scope.
func (m *Manager) SetKnowledgeProjectScope(fn func(context.Context, string, core.ID) (core.ID, error)) {
@@ -625,11 +670,17 @@ func (c *runCtx) Memories() []core.MemoryEntry {
}
func (c *runCtx) Emit(typ core.StreamEventType, payload any) error {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return err
}
_, err := c.mgr.emit(c.run.ConversationID, c.run.ID, typ, payload)
return err
}
func (c *runCtx) SetAssistantParts(parts []core.Part) error {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return err
}
c.mgr.mu.Lock()
c.h.assistant = parts
c.mgr.mu.Unlock()
@@ -637,6 +688,9 @@ func (c *runCtx) SetAssistantParts(parts []core.Part) error {
}
func (c *runCtx) AskHITL(card core.HITLCard) (answer core.HITLResponse, err error) {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.HITLResponse{}, err
}
card.ID = store.NewID("hitl")
card.RunID = c.run.ID
card.ConversationID = c.run.ConversationID
@@ -680,11 +734,14 @@ func (c *runCtx) AskHITL(card core.HITLCard) (answer core.HITLResponse, err erro
defer func() {
if acquireErr := c.mgr.adm.acquire(c.ctx, c.mgr.st, string(c.run.ID), c.run.Owner); acquireErr != nil {
log.Printf("[runmgr] admission reacquire after HITL %s: %v", c.run.ID, acquireErr)
if err == nil {
err = acquireErr
}
}
// Permission can change during both confirmation and admission waiting.
// Every HITL caller receives the live decision before it resumes work.
if err == nil {
err = c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID)
err = c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID)
}
}()
return waitHITL(c.ctx, c.h.hitlCh, timeout)
@@ -731,6 +788,9 @@ func (c *runCtx) Subhint() string { return c.subHint }
func (c *runCtx) Submodel() string { return c.subModel }
func (c *runCtx) UpdatePlan(p core.PlanSnapshot) error {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return err
}
prev, _ := c.mgr.st.GetPlan(c.run.ConversationID)
p.PlanID = c.run.ConversationID
if err := c.mgr.st.SavePlan(c.run.ConversationID, p); err != nil {
@@ -762,6 +822,9 @@ func (c *runCtx) Plan() (core.PlanSnapshot, bool) {
}
func (c *runCtx) UpdateTodos(l core.TodoList) error {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return err
}
prev, _ := c.mgr.st.GetTodos(c.run.ConversationID)
l.ConversationID = c.run.ConversationID
if err := c.mgr.st.SaveTodos(c.run.ConversationID, l); err != nil {
@@ -780,6 +843,9 @@ func (c *runCtx) UpdateTodos(l core.TodoList) error {
// (scope=project 工具参数),补全 ProjectID、ConversationID 保持空;其余照旧
// 归属当前会话。
func (c *runCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ActivityCard{}, err
}
if c.mgr.cards == nil {
return card, nil
}
@@ -792,6 +858,9 @@ func (c *runCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
}
func (c *runCtx) SetCardStatus(id string, status core.CardStatus, tone, note string) error {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return err
}
if c.mgr.cards == nil {
return nil
}
@@ -831,6 +900,9 @@ func (c *runCtx) BeginStreamCard(card core.ActivityCard) (core.ActivityCard, err
// AppendStreamCard:card_append 工具入口——向活跃流式卡追加正文(事件节流)。
func (c *runCtx) AppendStreamCard(text string) error {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return err
}
c.mgr.mu.Lock()
id := c.h.streamCardID
c.mgr.mu.Unlock()
@@ -843,6 +915,9 @@ func (c *runCtx) AppendStreamCard(text string) error {
// CommitStreamCard:card_commit 工具入口——活跃流式卡置 done、登记锚点回填
// 并清空活跃指针。
func (c *runCtx) CommitStreamCard() (core.ActivityCard, error) {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ActivityCard{}, err
}
c.mgr.mu.Lock()
id := c.h.streamCardID
c.h.streamCardID = ""
@@ -887,7 +962,7 @@ func (c *runCtx) Artifact(id string) (core.Artifact, []byte, bool) {
}
func (c *runCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.Artifact{}, err
}
art := core.Artifact{ProjectID: c.run.ProjectID, Owner: c.run.Owner, Name: name, Kind: kind, MIME: mime, SourceRun: c.run.ID}
@@ -999,7 +1074,7 @@ func (c *runCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
// ExecuteTool 执行画像策略:deny→错误;confirm→HITL confirm 卡片;执行→截断转工件。
func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()}
}
policy := c.prof.Tools
@@ -1022,7 +1097,7 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
if err != nil {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "hitl aborted: " + err.Error()}
}
if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()}
}
if resp.ChoiceID == "session" {
@@ -1031,7 +1106,7 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: "用户取消了本次工具调用"}
}
}
if err := c.mgr.authorizeProjectWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()}
}
// EXT-14:命令即工具——命中命令投影时经命令总分发(Entrance=agent、
@@ -1049,6 +1124,9 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
ID: call.Name, Params: params, Entrance: "agent",
Actor: c.run.Owner, Owner: c.run.Owner, ConversationID: c.run.ConversationID,
})
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()}
}
out, _ := json.Marshal(resp.Payload)
if resp.Card && resp.State == "accepted" {
return core.ToolCallResult{CallID: call.CallID, Status: "ok",
@@ -1063,6 +1141,9 @@ func (c *runCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
}
tc := &toolCtx{rc: c, callID: call.CallID}
res := c.mgr.registry.Execute(tc, call)
if err := c.mgr.authorizeRunWork(c.ctx, c.run.Owner, c.run.ProjectID); err != nil {
return core.ToolCallResult{CallID: call.CallID, Status: "error", Output: err.Error()}
}
// 工具产物回填(交付物卡片根因修复):注册表工具执行期经 rc.WriteArtifact
// 写入的工件(write_artifact / fs_write_file / register_artifact / 未来执行
// 原语产物)挂到结果上——llm.go 依 res.Artifact 生成对话内 artifact_ref 卡片。
@@ -1232,7 +1313,7 @@ func selectConversationRunner(m *Manager, eff core.EffectiveSettings, launch *wo
// attachments 为已解析的附件文档(server 层经 ingest 处理)。
// inputs 是同一批附件的原件,落到会话工作区 inputs/,供模板和仿制使用。
func (m *Manager) StartConversationRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) {
if err := m.authorizeProjectWork(context.Background(), conv.Owner, conv.ProjectID); err != nil {
if err := m.authorizeRunWork(context.Background(), conv.Owner, conv.ProjectID); err != nil {
return core.Run{}, err
}
if err := m.claimConversation(conv.ID); err != nil {
@@ -1244,7 +1325,7 @@ func (m *Manager) StartConversationRun(conv core.Conversation, userMessage core.
// StartWorkflowRun 启动一条绑定工作流的会话 run。定义在这里拷贝,
// 之后插件卸载不会改到已经在跑的这份。
func (m *Manager) StartWorkflowRun(conv core.Conversation, userMessage core.Message, def core.WorkflowDef, slots map[string]string, missing []workflow.SlotQuestion) (core.Run, error) {
if err := m.authorizeProjectWork(context.Background(), conv.Owner, conv.ProjectID); err != nil {
if err := m.authorizeRunWork(context.Background(), conv.Owner, conv.ProjectID); err != nil {
return core.Run{}, err
}
if err := m.claimConversation(conv.ID); err != nil {
@@ -1356,7 +1437,7 @@ func (m *Manager) claimConversation(conversationID string) error {
}
func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile, launch *workflow.Launch) (core.Run, error) {
if err := m.authorizeProjectWork(context.Background(), conv.Owner, conv.ProjectID); err != nil {
if err := m.authorizeRunWork(context.Background(), conv.Owner, conv.ProjectID); err != nil {
m.releaseConversation(conv.ID)
return core.Run{}, err
}
@@ -1371,7 +1452,7 @@ func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Messa
return run, err
}
ctx, cancel := context.WithCancel(context.Background())
h := &runHandle{cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
h := &runHandle{owner: run.Owner, cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
m.mu.Lock()
m.active[run.ID] = h
m.mu.Unlock()
@@ -1432,6 +1513,10 @@ func (m *Manager) execute(rc *runCtx, runner runtime.Runner) {
m.mu.Unlock()
m.deliverNotices(rc.run.ConversationID)
}()
if err := m.authorizeRunWork(rc.ctx, rc.run.Owner, rc.run.ProjectID); err != nil {
m.finishRun(rc, core.RunFailed, core.Usage{}, &core.ErrorPart{Code: "run.account_inactive", Message: err.Error()})
return
}
m.setRunStatus(&rc.run, core.RunRunning, core.PhaseThinking)
m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseThinking})
@@ -1457,6 +1542,14 @@ func (m *Manager) execute(rc *runCtx, runner runtime.Runner) {
}
func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage, errPart *core.ErrorPart) {
accountDenied := m.authorizeAccountEffective(rc.ctx, rc.run.Owner)
if accountDenied != nil {
status = core.RunFailed
errPart = &core.ErrorPart{Code: "run.account_inactive", Message: accountDenied.Error()}
if rc.h.cancel != nil {
rc.h.cancel()
}
}
m.mu.Lock()
usage.PromptTokens = rc.h.promptTokens
usage.CompletionTokens = rc.h.completionTokens
@@ -1481,6 +1574,9 @@ func (m *Manager) finishRun(rc *runCtx, status core.RunStatus, usage core.Usage,
m.mu.Lock()
parts := append([]core.Part{}, rc.h.assistant...)
m.mu.Unlock()
if accountDenied != nil {
parts = nil
}
if errPart != nil {
parts = append(parts, core.Part{Type: core.PartError, Err: errPart})
}
@@ -1685,7 +1781,7 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
if m.projectWorkAuthorization != nil && (run.Owner != owner || run.ConversationID != card.ConversationID) {
return "", false, store.ErrNotFound
}
if err := m.authorizeProjectWork(context.Background(), run.Owner, run.ProjectID); err != nil {
if err := m.authorizeRunWork(context.Background(), run.Owner, run.ProjectID); err != nil {
return "", false, err
}
// U7 双进程验收实测缺陷修复:run 不在本进程执行时不得先落 resolve——
@@ -1732,7 +1828,7 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
}
}
if pivot {
if err := m.authorizeProjectWork(context.Background(), run.Owner, run.ProjectID); err != nil {
if err := m.authorizeRunWork(context.Background(), run.Owner, run.ProjectID); err != nil {
return "", false, err
}
_ = m.CancelRun(run.ID)
@@ -1745,7 +1841,7 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
return trimmed, true, nil
}
}
if err := m.authorizeProjectWork(context.Background(), run.Owner, run.ProjectID); err != nil {
if err := m.authorizeRunWork(context.Background(), run.Owner, run.ProjectID); err != nil {
return "", false, err
}
resolution := core.HITLResolvedPayload{CardID: core.ID(cardID), Answer: answer, Kind: kind}
@@ -1780,6 +1876,9 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
// ---- 子代理(任务树)----
func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agentID, name, task, hint string) (core.AgentResultPayload, error) {
if err := m.authorizeRunWork(parentCtx, parent.Owner, parent.ProjectID); err != nil {
return core.AgentResultPayload{}, err
}
rc, runner, err := m.startChild(parentCtx, parent, agentID, name, task, hint, false)
if err != nil {
return core.AgentResultPayload{}, err
@@ -1811,6 +1910,9 @@ func (m *Manager) startChild(parentCtx context.Context, parent core.Run, agentID
// startChildOpts 在 startChild 之上支持系统派发选项(模型档等)。
func (m *Manager) startChildOpts(parentCtx context.Context, parent core.Run, o bgSpawnOpts, detach bool) (*runCtx, runtime.Runner, error) {
if err := m.authorizeRunWork(parentCtx, parent.Owner, parent.ProjectID); err != nil {
return nil, nil, err
}
agentID, name, task, hint := o.agentID, o.name, o.task, o.hint
if parent.Kind == core.RunSubagent {
return nil, nil, fmt.Errorf("子代理不能再派发子代理")
@@ -1835,7 +1937,7 @@ func (m *Manager) startChildOpts(parentCtx context.Context, parent core.Run, o b
base = context.Background()
}
ctx, cancel := context.WithCancel(base)
h := &runHandle{cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
h := &runHandle{owner: child.Owner, cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
m.mu.Lock()
m.active[child.ID] = h
m.mu.Unlock()
@@ -1855,14 +1957,24 @@ func (m *Manager) startChildOpts(parentCtx context.Context, parent core.Run, o b
// finishChild 执行子 run、落库并汇总。emitResult 为假时不发 agent.result,
// 留给后台投递成功之后再发。
func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool) (core.AgentResultPayload, error) {
execErr := func() (err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("subagent panic: %v", r)
}
execErr := m.authorizeRunWork(rc.ctx, rc.run.Owner, rc.run.ProjectID)
if execErr == nil {
execErr = func() (err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("subagent panic: %v", r)
}
}()
return runner.Execute(rc)
}()
return runner.Execute(rc)
}()
}
accountDenied := m.authorizeAccountEffective(rc.ctx, rc.run.Owner)
if accountDenied != nil {
execErr = accountDenied
if rc.h.cancel != nil {
rc.h.cancel()
}
}
// 子代理记账(N3 WP6):token 用量随子 run 入账(此前只有 DurationMs)。
usage := core.Usage{DurationMs: float64(time.Since(rc.h.started).Milliseconds())}
@@ -1879,7 +1991,11 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool
summary := "完成"
payload := core.AgentResultPayload{AgentRun: rc.run.ID, Status: status, Summary: summary}
if execErr != nil {
if rc.ctx.Err() != nil {
if accountDenied != nil {
status = core.RunFailed
payload.Status = status
payload.Summary = accountDenied.Error()
} else if rc.ctx.Err() != nil {
status = core.RunCancelled
payload.Status = status
payload.Summary = "已取消"
@@ -1896,13 +2012,16 @@ func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool
child.Phase = core.PhaseDone
child.Usage = &usage
m.st.UpdateRun(&child)
if emitResult {
if emitResult && accountDenied == nil {
m.emit(child.ConversationID, child.ID, core.EvAgentResult, payload)
}
m.mu.Lock()
subParts := append([]core.Part{}, rc.h.assistant...)
m.mu.Unlock()
if accountDenied != nil {
subParts = nil
}
// 独立转录落定(含终态 done 标记,供续跑幂等判定)。
m.writeSubagentTranscript(child, subParts, string(status))
fullText := partsText(subParts)
@@ -1944,6 +2063,9 @@ func (m *Manager) spawnBackground(parent core.Run, agentID, name, task, hint str
}
func (m *Manager) spawnBackgroundOpts(parent core.Run, o bgSpawnOpts) (core.ID, error) {
if err := m.authorizeRunWork(context.Background(), parent.Owner, parent.ProjectID); err != nil {
return "", err
}
if parent.Kind == core.RunSubagent {
return "", fmt.Errorf("子代理不能再派发子代理")
}
@@ -1990,6 +2112,10 @@ func (m *Manager) spawnBackgroundOpts(parent core.Run, o bgSpawnOpts) (core.ID,
// startBackgroundSlot 持槽启动后台任务(调用方已占 bgInFlight)。
// preCard 非空时复用排队卡(回填 SourceRunID 并转执行态),否则新建执行卡。
func (m *Manager) startBackgroundSlot(parent core.Run, o bgSpawnOpts, preCard core.ActivityCard) (core.ID, error) {
if err := m.authorizeRunWork(context.Background(), parent.Owner, parent.ProjectID); err != nil {
m.decBackground(parent.ConversationID)
return "", err
}
rc, runner, err := m.startChildOpts(context.Background(), parent, o, true)
if err != nil {
m.decBackground(parent.ConversationID)
@@ -2271,6 +2397,10 @@ func (m *Manager) startNoticeRun(convID string, items []bgNotice) error {
m.releaseConversation(convID)
return err
}
if err := m.authorizeRunWork(context.Background(), owner, conv.ProjectID); err != nil {
m.releaseConversation(convID)
return err
}
var b strings.Builder
b.WriteString("<system-reminder>\n")
for _, n := range items {
@@ -2300,6 +2430,10 @@ func (m *Manager) startNoticeRun(convID string, items []bgNotice) error {
ConversationID: convID, Role: core.RoleSystem,
Parts: []core.Part{{Type: core.PartText, Text: b.String()}},
}
if err := m.authorizeRunWork(context.Background(), owner, conv.ProjectID); err != nil {
m.releaseConversation(convID)
return err
}
if err := m.st.AppendMessage(&msg); err != nil {
m.releaseConversation(convID)
return err