Bind formal commands to exact project sources

Why: Chat intent, cross-conversation anchors, payload changes, and native workspace tools could bypass project source analysis.

Evidence: Command-bus, app, server, and squad-adapter suites pass with centralized preflight and project-tool fail-closed behavior.

Impact: Every classified formal entrance binds actor, conversation, project, command, payload digest, and source watermark before dispatch.

Refs: TT-N03 TT-R01
This commit is contained in:
weichen
2026-10-15 07:00:18 +00:00
parent f8ea71f6b1
commit 14e780c6fa
9 changed files with 84 additions and 59 deletions
+25 -4
View File
@@ -5,6 +5,7 @@ package app
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
@@ -412,7 +413,7 @@ func Run(opt Options) error {
cfg.Fetch = nil
cfg.MCP = nil
}
cfg.CodeOrchestration = effNow.GetBool("runtime.codeOrchestration", false)
cfg.CodeOrchestration = rc.Run().ProjectID == "" && effNow.GetBool("runtime.codeOrchestration", false)
if ask, ok := rc.(interface {
AskUser(card core.HITLCard) (core.HITLResponse, error)
}); ok {
@@ -641,6 +642,29 @@ func Run(opt Options) error {
}
return nil
}}
bus.SetProjectWorkPreflight(func(ctx context.Context, req cmdbus.Request) error {
if req.Actor == "" || req.Actor != req.Owner || req.ConversationID == "" || req.AnchorMessageID == "" || req.HostProjectID == "" {
return fmt.Errorf("formal project command requires an authenticated conversation source")
}
conv, err := st.GetConversation(req.Owner, req.ConversationID)
if err != nil || string(conv.ProjectID) != req.HostProjectID {
return fmt.Errorf("formal project command conversation binding changed")
}
raw, err := json.Marshal(req.Params)
if err != nil {
return err
}
digest := sha256.Sum256(raw)
decision, err := preflightProjectMessage(ctx, st, workObserverSlot, projectSlot.work(), req.AnchorMessageID,
"project-command:"+req.ID, req.Owner, req.ConversationID, req.HostProjectID, req.ID, hex.EncodeToString(digest[:]))
if err != nil {
return err
}
if decision.State != extension.ProjectWorkProceed || decision.DecisionID == "" {
return fmt.Errorf("source analysis did not commit a proceed decision")
}
return nil
})
// EXT-13:回合内检索(knowledge_search / @knowledge 注入)与 HTTP/Host 检索面
// 同一适配器授权;槽为共享指针,扩展在 applyCompiled 期间注册后即生效。
mgr.SetSharedKnowledgeAllow(func(owner string, projectID core.ID) bool {
@@ -707,9 +731,6 @@ func Run(opt Options) error {
}
srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost)
srv.SetProjectCommandPreflight(func(ctx context.Context, messageID, action string) (extension.ProjectWorkDecision, error) {
return preflightProjectMessage(ctx, st, workObserverSlot, projectSlot.work(), messageID, action)
})
projectSlot.wire(srv, caps)
// EXT-8:装载期注册的资源授权适配器注入读取入口(未注册=默认拒绝,行为不变)。
srv.SetProjectAuthorizer(authzSlot.Authorize)
+5 -2
View File
@@ -224,7 +224,7 @@ func verifyProjectTransientObservation(_ context.Context, st *store.Store, row s
// inline, then asks RDM to prove that every exact source part was analyzed for
// the pending action. A durable intake ACK alone never permits the action.
func preflightProjectMessage(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot,
work func(context.Context, string, core.ID) (bool, error), messageID, action string,
work func(context.Context, string, core.ID) (bool, error), messageID, action, expectedOwner, expectedConversation, expectedProject, targetID, targetVersion string,
) (extension.ProjectWorkDecision, error) {
pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "project source awaiting analysis"}
provider := slot.get()
@@ -256,6 +256,9 @@ func preflightProjectMessage(ctx context.Context, st *store.Store, slot *project
})
}
first := rows[0]
if first.Owner != expectedOwner || first.ConversationID != expectedConversation || first.ProjectID != expectedProject {
return pending, errors.New("extension: source message is not bound to this actor, conversation, and project")
}
for _, row := range rows[1:] {
if row.Owner != first.Owner || row.ProjectID != first.ProjectID || row.ConversationID != first.ConversationID {
return pending, errors.New("extension: mixed project source manifest")
@@ -276,7 +279,7 @@ func preflightProjectMessage(ctx context.Context, st *store.Store, slot *project
watermark := fmt.Sprintf("%x", sha256.Sum256(raw))
decision, err := provider.Preflight(ctx, extension.ProjectWorkAction{
Actor: first.Owner, Owner: first.Owner, ProjectID: conv.ProjectID,
ConversationID: conv.ID, RunID: core.ID(first.RunID), Action: action,
ConversationID: conv.ID, RunID: core.ID(first.RunID), Action: action, TargetID: targetID, TargetVersion: targetVersion,
Manifest: manifest, SourceWatermark: watermark,
})
if err != nil {
@@ -116,19 +116,19 @@ func TestProjectMessagePreflightRequiresAnalysisAndExactManifest(t *testing.T) {
work := func(_ context.Context, actor string, id core.ID) (bool, error) {
return actor == "alice" && id == project.ID, nil
}
decision, err := preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run")
decision, err := preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
if err != nil || decision.State != extension.ProjectWorkPending || observations != 1 || decisions != 1 {
t.Fatalf("durable intake was confused with completed analysis: %+v observations=%d decisions=%d err=%v", decision, observations, decisions, err)
}
ready = true
decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run")
decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
if err != nil || decision.State != extension.ProjectWorkProceed || observations != 1 || decisions != 2 {
t.Fatalf("same source did not proceed after domain analysis: %+v observations=%d decisions=%d err=%v", decision, observations, decisions, err)
}
if _, err := st.DB().Exec(`UPDATE messages SET parts=? WHERE id=?`, `[{"type":"text","text":"changed"}]`, msg.ID); err != nil {
t.Fatal(err)
}
decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run")
decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
if err != nil || decision.State != extension.ProjectWorkPending || decisions != 2 {
t.Fatalf("source changed after ACK but stale decision still passed: %+v decisions=%d err=%v", decision, decisions, err)
}
@@ -136,7 +136,7 @@ func TestProjectMessagePreflightRequiresAnalysisAndExactManifest(t *testing.T) {
if err := st.AppendMessageWithRequestAccountChecked("alice", false, &partial, "req-2"); err != nil {
t.Fatal(err)
}
decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(partial.ID), "conversation.run")
decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(partial.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
if err != nil || decision.State != extension.ProjectWorkPending || decisions != 2 {
t.Fatalf("partial attachment reached project action preflight: %+v decisions=%d err=%v", decision, decisions, err)
}
+30 -15
View File
@@ -124,7 +124,9 @@ func RequiresProjectWorkDecision(id string) bool {
}
}
switch id {
case "rdm.structure.propose", "rdm.structure.revise", "rdm.docs.instantiate", "rdm.ingest", "rdm.upload.confirm", "rdm.knowledge.publish", "rdm.trace.confirm":
case "rdm.structure.propose", "rdm.structure.revise", "rdm.docs.instantiate", "rdm.docs.delivery.approve",
"rdm.doc.snapshot", "rdm.ingest", "rdm.upload.confirm", "rdm.knowledge.publish", "rdm.trace.confirm",
"rdm.conflicts.resolve":
return true
default:
return false
@@ -154,20 +156,25 @@ type pending struct {
}
type Bus struct {
reg *agentcommand.Registry
audit AuditStore
cards CardPort
now func() time.Time
handlers map[string]Handler
slash map[string]string // "/x" → command id
pendMu sync.Mutex
pending map[string]pending
pendingStore PendingStore
intents *IntentStore // G2-2:意图声明表(自然语言路由)
toolGate ToolGate // U3:owner 级工具注入闸(nil = 不过滤)
accountGuard EffectiveAccountGuard // A1: optional command execution precheck
strictAccountWrites bool
allowImplicitLocal bool
reg *agentcommand.Registry
audit AuditStore
cards CardPort
now func() time.Time
handlers map[string]Handler
slash map[string]string // "/x" → command id
pendMu sync.Mutex
pending map[string]pending
pendingStore PendingStore
intents *IntentStore // G2-2:意图声明表(自然语言路由)
toolGate ToolGate // U3:owner 级工具注入闸(nil = 不过滤)
accountGuard EffectiveAccountGuard // A1: optional command execution precheck
strictAccountWrites bool
allowImplicitLocal bool
projectWorkPreflight func(context.Context, Request) error
}
func (b *Bus) SetProjectWorkPreflight(fn func(context.Context, Request) error) {
b.projectWorkPreflight = fn
}
// PendingStore 把危险命令的确认挂起态落盘。进程内 map 仍是热路径;重启后从存储取回。
@@ -375,6 +382,14 @@ func (b *Bus) Dispatch(ctx context.Context, req Request) Response {
if err := validateParams(spec.ParamsSchema, req.Params); err != nil {
return receipt(agentcommand.StateRejected, "", "params: "+err.Error())
}
if RequiresProjectWorkDecision(req.ID) {
if b.projectWorkPreflight == nil {
return receipt(agentcommand.StateRejected, "", "project source preflight unavailable")
}
if err := b.projectWorkPreflight(ctx, req); err != nil {
return receipt(agentcommand.StateRejected, "", "project source preflight: "+err.Error())
}
}
// 高危收口:非确认入口先落内联确认卡(committed 语义 = 卡已创建;确认后二次分发执行)。
if spec.Danger == agentcommand.DangerHigh && !req.Confirmed {
if b.cards == nil || req.ConversationID == "" {
+2
View File
@@ -62,6 +62,7 @@ func newTestBus(a AuditStore, c CardPort) *Bus {
func TestDispatchFourEntrancesSameAudit(t *testing.T) {
a := &memAudit{}
b := newTestBus(a, &memCards{})
b.SetProjectWorkPreflight(func(context.Context, Request) error { return nil })
for _, ent := range []string{"button", "slash", "chat", "card", "api", "weird-unknown"} {
resp := b.Dispatch(context.Background(), Request{ID: "demo.ping", Entrance: ent})
if resp.Receipt.State != agentcommand.StateCommitted {
@@ -277,6 +278,7 @@ func TestValidateParamsScalars(t *testing.T) {
func TestRegisterPluginCommandDispatch(t *testing.T) {
a := &memAudit{}
b := newTestBus(a, &memCards{})
b.SetProjectWorkPreflight(func(context.Context, Request) error { return nil })
err := b.RegisterPluginCommand(agentcommand.CommandSpec{ID: "rdm.ingest", Title: "入库"}, func(ctx context.Context, req Request) (any, error) {
return map[string]any{"echo": req.Params["ref"], "owner": req.Owner}, nil
})
+1 -1
View File
@@ -3,7 +3,7 @@ package cmdbus
import "testing"
func TestFormalRDMWritesRequireProjectWorkDecision(t *testing.T) {
for _, id := range []string{"rdm.reqs.add", "rdm.docs.instantiate", "rdm.review.doc.approve", "rdm.structure.propose", "rdm.trace.confirm"} {
for _, id := range []string{"rdm.reqs.add", "rdm.docs.instantiate", "rdm.docs.delivery.approve", "rdm.doc.snapshot", "rdm.conflicts.resolve", "rdm.review.doc.approve", "rdm.structure.propose", "rdm.trace.confirm"} {
if !RequiresProjectWorkDecision(id) {
t.Errorf("formal write %s bypassed project source preflight", id)
}
+1
View File
@@ -67,6 +67,7 @@ func TestIntentDeclareMatchRoute(t *testing.T) {
t.Fatalf("匹配 = %+v ok=%v", in, ok)
}
// 路由:匹配到的命令经总线 chat 入口分发成功。
bus.SetProjectWorkPreflight(func(context.Context, Request) error { return nil })
resp := bus.Dispatch(context.Background(), Request{ID: in.Command, Params: in.Params, Entrance: "chat", Owner: "alice"})
if resp.Receipt.State != "committed" {
t.Fatalf("路由分发 = %+v", resp.Receipt)
+4 -26
View File
@@ -79,8 +79,7 @@ type Server struct {
// 全部动作权威;未注册时 read 回落 ProjectAuthz(兼容),其余动作一律
// 默认拒绝——旧只读适配器不得自动获得写入能力。公开契约的版本化扩展
// 计划见 docs/guides/team-storage.md(KTD13 一节)。
ProjectAuthzActions func(owner string, projectID core.ID, action ProjectAction) (bool, error)
ProjectCommandPreflight func(context.Context, string, string) (extension.ProjectWorkDecision, error)
ProjectAuthzActions func(owner string, projectID core.ID, action ProjectAction) (bool, error)
// CompiledPlugins 编译进二进制的进程内插件摘要(组合根在 applyCompiled
// 期间记录注册面生成;nil/空 = 无编译插件,清单行为不变)。
CompiledPlugins []CompiledPlugin
@@ -114,10 +113,6 @@ func (s *Server) SetProjectActionAuthorizer(fn func(owner string, projectID core
s.ProjectAuthzActions = fn
}
func (s *Server) SetProjectCommandPreflight(fn func(context.Context, string, string) (extension.ProjectWorkDecision, error)) {
s.ProjectCommandPreflight = fn
}
// authorizeProjectAction 动作矩阵唯一判定(KTD13):
// - action-aware 适配器已注册 → 权威(出错/拒绝一律 false,fail-closed);
// - 未注册 → read 回落旧只读 ProjectAuthz(兼容 EXT-8 现网),其余动作
@@ -280,7 +275,8 @@ func handleSystemIntent(ctx context.Context, b *cmdbus.Bus, req cmdbus.Request)
}
}
resp := b.Dispatch(ctx, cmdbus.Request{ID: in.Command, Params: params, Entrance: "chat",
Actor: req.Actor, Owner: req.Owner, ConversationID: req.ConversationID})
Actor: req.Actor, Owner: req.Owner, ConversationID: req.ConversationID,
AnchorMessageID: req.AnchorMessageID, HostProjectID: req.HostProjectID})
return map[string]any{"matched": true, "intent": in.ID, "command": in.Command,
"dispatched": true, "receipt": resp.Receipt, "payload": resp.Payload}, nil
}
@@ -1151,7 +1147,7 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
if routeKind == route.KindCommand {
resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{
ID: recognized.ID, Params: map[string]any{"text": text}, Entrance: "chat", Actor: owner,
Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID,
Owner: owner, ConversationID: convID, AnchorMessageID: msg.ID, HostProjectID: string(conv.ProjectID),
})
out := map[string]any{"message": msg, "receipt": resp.Receipt, "command": recognized.ID}
if resp.Card != nil {
@@ -1286,24 +1282,6 @@ func (s *Server) execCommand(w http.ResponseWriter, r *http.Request) {
body.Params = map[string]any{}
}
commandID := r.PathValue("id")
if cmdbus.RequiresProjectWorkDecision(commandID) {
if body.ConversationID == "" || body.AnchorMessageID == "" || s.ProjectCommandPreflight == nil {
http.Error(w, "project command requires a committed source message", http.StatusConflict)
return
}
decision, err := s.ProjectCommandPreflight(r.Context(), body.AnchorMessageID, "project-command:"+commandID)
if err != nil || decision.State != extension.ProjectWorkProceed || decision.DecisionID == "" {
reason := decision.Reason
if err != nil {
reason = err.Error()
}
if reason == "" {
reason = "source analysis did not commit a proceed decision"
}
http.Error(w, "project command source analysis is not complete: "+reason, http.StatusConflict)
return
}
}
var hostProject string
if body.ConversationID != "" {
conv, err := s.st.GetConversation(s.owner(r), body.ConversationID)
+12 -7
View File
@@ -75,9 +75,10 @@ func RunEntry(rc Ctx, opt EntryOptions) (EntryResult, error) {
hostNames = append(hostNames, tool.Name())
}
registry := squad.NewRoleRegistry()
projectRun := rc.Run().ProjectID != ""
for _, role := range kernelRoles(roleCaps{
host: hostNames, exec: cfg.Exec != nil, skills: len(cfg.SkillsRoots) > 0,
search: cfg.Search != nil, fetch: cfg.Fetch != nil,
search: cfg.Search != nil, fetch: cfg.Fetch != nil, workspace: !projectRun,
}) {
if err := registry.Register(role); err != nil {
return EntryResult{}, fmt.Errorf("squad entry: register %s: %w", role.Role, err)
@@ -167,15 +168,19 @@ func readDeliverable(res *squad.RunResult) string {
}
type roleCaps struct {
host []string
exec bool
skills bool
search bool
fetch bool
host []string
exec bool
skills bool
search bool
fetch bool
workspace bool
}
func kernelRoles(caps roleCaps) []squad.RoleSpec {
base := []string{"read_file", "write_file", "list_dir", "grep"}
base := []string{}
if caps.workspace {
base = []string{"read_file", "write_file", "list_dir", "grep"}
}
writerTools := append([]string{}, base...)
writerTools = append(writerTools, "edit_file", "multi_edit", "glob")
if caps.exec {