From d2efea284521bc68e775a7483dfbe321b6adf97e Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 03:20:41 +0000 Subject: [PATCH] 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 --- internal/app/app.go | 13 +- internal/runmgr/account_guard_test.go | 145 ++++++++++++++++++++ internal/runmgr/runmgr.go | 186 ++++++++++++++++++++++---- 3 files changed, 317 insertions(+), 27 deletions(-) create mode 100644 internal/runmgr/account_guard_test.go diff --git a/internal/app/app.go b/internal/app/app.go index 33749b7..cbc202b 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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 )") diff --git a/internal/runmgr/account_guard_test.go b/internal/runmgr/account_guard_test.go new file mode 100644 index 0000000..ef363c7 --- /dev/null +++ b/internal/runmgr/account_guard_test.go @@ -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") + } + } + } +} diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index b3f3d99..0833ca5 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -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("\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