mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
Merge pull request #14394 from SivanCola/codex/optimistic-call-admission-14025
fix(agent): coordinate optimistic child writes / 修复乐观子代理具名写入绕过占用
This commit is contained in:
8 files changed
+411
-5
No files matched your search
@@ -0,0 +1,64 @@
|
||||
---
|
||||
owner: @SivanCola
|
||||
backup: @esengine
|
||||
status: active
|
||||
reviewed: 2026-10-25
|
||||
---
|
||||
|
||||
# Optimistic named writes / 乐观子任务具名写入
|
||||
|
||||
## Cause and owner / 根因与所有者
|
||||
|
||||
- The task registry skipped explicit-fence binding for whole-workspace claims,
|
||||
including optimistic claims. Child agents also lacked the shared scheduler;
|
||||
the parent-only admission explicitly skipped nonzero subagent depth.
|
||||
- A real background task could therefore own a file while an optimistic sibling
|
||||
wrote it through either write_file or a dynamically resolved local capability.
|
||||
|
||||
乐观任务没有安装显式路径围栏包装器,子代理也没有共享调度器的实际调用准入。
|
||||
真实后台任务持有文件分工时,乐观兄弟任务仍可通过直接或动态工具写入该文件。
|
||||
|
||||
## Repair and authority / 修复与权限
|
||||
|
||||
- The common child-options owner shares the scheduler. Concrete execution admits
|
||||
named writes after target resolution and permission, before pre-hooks and the
|
||||
actual call. The whole path set is reserved once and released by call cleanup.
|
||||
- The root comes from that actor's existing grant. WriteWorkspaceRoot is not
|
||||
enabled merely for normalization: it also changes default observation scope.
|
||||
- Parent admission and explicit child wrappers retain their existing owners.
|
||||
Existing wrappers are not double-reserved. Opaque legacy Bash/MCP/custom tools
|
||||
retain their policy and availability; the entire tool set is not fence-filtered.
|
||||
|
||||
统一子代理构造器传递同一调度器,在真实目标解析和权限检查后、hooks 和执行前申请完整
|
||||
具名路径集合,并由本次调用清理释放。路径根取自该执行主体已有授权,不为锁而扩大
|
||||
工作区观察范围。保留父代理、显式围栏和未知范围工具的既有分工与权限。
|
||||
|
||||
## Evidence / 证据
|
||||
|
||||
- Real boot.Build holds an explicit background task in its first model request.
|
||||
Task/fleet siblings use direct or capability writers on overlapping/disjoint
|
||||
files. Overlap is refused; disjoint writes finish while the holder is live.
|
||||
- The compiling original owners write the held file in both direct and dynamic
|
||||
task cases. The corrected execution owner refuses them. Unit probes also
|
||||
require all-or-nothing multi-path admission and release while the actor lives.
|
||||
- Real provider requests retain the same serialized tools and system prefix.
|
||||
Local and hosted platform/race evidence are recorded separately in the PR.
|
||||
|
||||
回归使用真实装配和通道控制后台模型等待,不靠瞬时竞争或私有字段伪造执行状态。
|
||||
直接和动态 task 反例已在旧所有者上编译并实际写入占用文件;修复拒绝冲突,保留不相交
|
||||
写入。另测多路径失败不留下部分占用、调用结束释放,以及实际模型前缀稳定。
|
||||
|
||||
## Limits and compatibility / 限制与兼容
|
||||
|
||||
- This is session-local legacy scheduling admission. It is not cross-process
|
||||
CAS, content-version enforcement, automatic isolation or the new transaction
|
||||
protocol. Older processes cannot gain these checks from this process.
|
||||
- Explicit task-wide reservations, workspace lease retention, opaque hook effects,
|
||||
unknown write extents and external writers remain separate review units.
|
||||
- No stored format, tool definition, wire type, permission grant, mode or default
|
||||
changes. Conflict uses the existing write.path_busy refusal and retry policy.
|
||||
- Author assessment does not replace independent code-owner/security approval.
|
||||
|
||||
本批只修复既有模式下同会话的具名写入准入,不宣称完成跨进程 CAS、强制版本检查或
|
||||
自动隔离。任务长预约、工作区长租约、未知 hooks/写入范围、外部进程仍待后续处理。
|
||||
没有新增持久字段、工具定义、权限或默认策略;作者分析不能替代独立审查。
|
||||
@@ -0,0 +1,190 @@
|
||||
package boot
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/session/control"
|
||||
)
|
||||
|
||||
type admissionHolderProvider struct {
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (*admissionHolderProvider) Name() string { return "boot-admission-holder" }
|
||||
|
||||
func (p *admissionHolderProvider) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Chunk, error) {
|
||||
p.once.Do(func() { close(p.started) })
|
||||
select {
|
||||
case <-p.release:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
ch := make(chan provider.Chunk, 2)
|
||||
ch <- provider.Chunk{Type: provider.ChunkText, Text: "finished without changes"}
|
||||
ch <- provider.Chunk{Type: provider.ChunkDone}
|
||||
close(ch)
|
||||
return ch, nil
|
||||
}
|
||||
|
||||
// A real background task owns an explicit path while its first model request
|
||||
// is blocked. The optimistic sibling must honor that live scheduler claim,
|
||||
// through either a direct file tool or a dynamically resolved capability.
|
||||
func TestEffectOptimisticSiblingHonorsNamedWriteAdmission(t *testing.T) {
|
||||
for _, dispatch := range []string{"task", "fleet"} {
|
||||
for _, dynamic := range []bool{false, true} {
|
||||
for _, overlap := range []bool{false, true} {
|
||||
mode, scope := "direct", "disjoint"
|
||||
if dynamic {
|
||||
mode = "capability"
|
||||
}
|
||||
if overlap {
|
||||
scope = "overlap"
|
||||
}
|
||||
t.Run(dispatch+"/"+mode+"/"+scope, func(t *testing.T) {
|
||||
home, dir := statedBootHome(t), robustTempDir(t)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
|
||||
defer cancel()
|
||||
holder := &admissionHolderProvider{started: make(chan struct{}), release: make(chan struct{})}
|
||||
observer := &browserScriptProvider{}
|
||||
target := filepath.Join(dir, "other.txt")
|
||||
if overlap {
|
||||
target = filepath.Join(dir, "claimed.txt")
|
||||
}
|
||||
worker := &browserScriptProvider{rounds: []func(string) *provider.ToolCall{
|
||||
func(string) *provider.ToolCall {
|
||||
args := map[string]any{"path": target, "content": "optimistic write\n"}
|
||||
if dynamic {
|
||||
return browserCall("writer", "use_capability", map[string]any{
|
||||
"action": "call", "capability_id": "tool:write_file", "arguments": args,
|
||||
})
|
||||
}
|
||||
return browserCall("writer", "write_file", args)
|
||||
},
|
||||
}}
|
||||
parent := &browserScriptProvider{rounds: []func(string) *provider.ToolCall{
|
||||
func(string) *provider.ToolCall {
|
||||
return browserCall("holder", "task", map[string]any{
|
||||
"prompt": "Hold claimed.txt until released", "model": "holder",
|
||||
"write_paths": []string{"claimed.txt"}, "run_in_background": true,
|
||||
})
|
||||
},
|
||||
func(string) *provider.ToolCall {
|
||||
select {
|
||||
case <-holder.started:
|
||||
case <-ctx.Done():
|
||||
t.Error("background task never entered its model request")
|
||||
return nil
|
||||
}
|
||||
if dispatch == "fleet" {
|
||||
return browserCall("optimistic", "fleet", map[string]any{"tasks": []any{
|
||||
map[string]any{"id": "writer", "prompt": "Write the requested file", "model": "worker"},
|
||||
map[string]any{"id": "observer", "prompt": "Observe without writing", "model": "observer", "read_only": true},
|
||||
}})
|
||||
}
|
||||
return browserCall("optimistic", dispatch, map[string]any{
|
||||
"prompt": "Write the requested file", "model": "worker",
|
||||
})
|
||||
},
|
||||
}}
|
||||
kind := "boot-optimistic-admission-" + strings.ToLower(t.Name())
|
||||
provider.Register(kind, func(cfg provider.Config) (provider.Provider, error) {
|
||||
switch cfg.Model {
|
||||
case "holder":
|
||||
return holder, nil
|
||||
case "worker":
|
||||
return worker, nil
|
||||
case "observer":
|
||||
return observer, nil
|
||||
default:
|
||||
return parent, nil
|
||||
}
|
||||
})
|
||||
writeFile(t, dir, "reasonix.toml", `
|
||||
default_model = "parent"
|
||||
|
||||
[agent]
|
||||
subagent_claims = "optimistic"
|
||||
|
||||
[codegraph]
|
||||
enabled = false
|
||||
|
||||
[[providers]]
|
||||
name = "parent"
|
||||
kind = "`+kind+`"
|
||||
model = "parent"
|
||||
|
||||
[[providers]]
|
||||
name = "holder"
|
||||
kind = "`+kind+`"
|
||||
model = "holder"
|
||||
|
||||
[[providers]]
|
||||
name = "worker"
|
||||
kind = "`+kind+`"
|
||||
model = "worker"
|
||||
|
||||
[[providers]]
|
||||
name = "observer"
|
||||
kind = "`+kind+`"
|
||||
model = "observer"
|
||||
`)
|
||||
approveWorkspaceAt(t, home, dir)
|
||||
ctrl, err := Build(ctx, Options{Home: home, WorkspaceRoot: dir, Sink: event.Discard})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
close(holder.release)
|
||||
ctrl.Close()
|
||||
if manager := ctrl.JobManager(); manager != nil {
|
||||
done := make(chan struct{})
|
||||
manager.AfterUnfinished(func() { close(done) })
|
||||
<-done
|
||||
}
|
||||
})
|
||||
ctrl.SetToolApprovalMode(control.ToolApprovalYolo)
|
||||
if err := ctrl.Run(ctx, "Start a background holder, then run the optimistic writer"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reqs := worker.requests()
|
||||
if len(reqs) < 2 {
|
||||
t.Fatalf("optimistic writer never returned a tool result: %d requests", len(reqs))
|
||||
}
|
||||
results := effectToolResults(reqs[len(reqs)-1])
|
||||
if len(results) != 1 {
|
||||
t.Fatalf("writer results = %q", results)
|
||||
}
|
||||
content, readErr := os.ReadFile(target)
|
||||
if overlap {
|
||||
if !os.IsNotExist(readErr) || !strings.Contains(results[0], "blocked:") {
|
||||
t.Fatalf("optimistic writer bypassed the live claim: file=%q, read=%v, result=%q", content, readErr, results[0])
|
||||
}
|
||||
} else if readErr != nil || string(content) != "optimistic write\n" {
|
||||
t.Fatalf("disjoint writer was blocked: file=%q, read=%v, result=%q", content, readErr, results[0])
|
||||
}
|
||||
for _, requests := range [][]provider.Request{parent.requests(), reqs} {
|
||||
first, _ := json.Marshal(requests[0].Tools)
|
||||
for _, request := range requests[1:] {
|
||||
tools, _ := json.Marshal(request.Tools)
|
||||
if !bytes.Equal(first, tools) || systemOf(request) != systemOf(requests[0]) {
|
||||
t.Fatal("admission changed the stable provider prefix")
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -128,10 +128,9 @@ type Options struct {
|
||||
// queue. When nil, child construction shadows inherited queues.
|
||||
MemoryQueue memory.Queue
|
||||
|
||||
// WriteScheduler is the session-scoped subagent concurrency/write-claim
|
||||
// controller. When set on the parent executor, write-capable tools reserve
|
||||
// paths for the duration of Execute so background writers cannot TOCTOU
|
||||
// race parent writes. Subagents leave this nil (or depth > 0 skips it).
|
||||
// WriteScheduler coordinates concrete writes with delegated claims. The parent
|
||||
// and optimistic children share it; explicitly fenced children keep their
|
||||
// existing path-bound wrapper owner.
|
||||
WriteScheduler *writeclaim.SubagentScheduler
|
||||
// WriteWorkspaceRoot normalizes parent write reservations.
|
||||
WriteWorkspaceRoot string
|
||||
|
||||
@@ -578,7 +578,7 @@ func (a *Agent) prepareToolExecution(ctx context.Context, plan *toolCallPlan) (t
|
||||
// check-before-write TOCTOU windows. Dynamic Economy/MCP tools are covered
|
||||
// here after registry lookup without schema-changing wrappers.
|
||||
// executeOne defers plan.releaseParentWrite so every return path releases.
|
||||
if releaseParentWrite, perr := a.reserveParentWrite(plan.runTool, plan.runArgs, plan.readOnly, a.declareWritePaths(ctx, plan)); perr != nil {
|
||||
if releaseParentWrite, perr := a.reserveToolWrite(ctx, plan.runTool, plan.runArgs, plan.readOnly, a.declareWritePaths(ctx, plan)); perr != nil {
|
||||
out := toolOutcome{
|
||||
output: "blocked: " + perr.Error(),
|
||||
blocked: true,
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/runtime/writeclaim"
|
||||
)
|
||||
|
||||
// reserveToolWrite admits the actual execution target, including a child proxy's
|
||||
// resolved file writer. Explicit child fences keep their existing wrapper owner;
|
||||
// legacy opaque effects keep their existing policy until effect coordination.
|
||||
func (a *Agent) reserveToolWrite(ctx context.Context, writer tool.Tool, args json.RawMessage, readOnly bool, declared []string) (func(), error) {
|
||||
if a == nil || a.subagentDepth <= 0 {
|
||||
return a.reserveParentWrite(writer, args, readOnly, declared)
|
||||
}
|
||||
noop := func() {}
|
||||
if readOnly || writer == nil || a.svc.writeScheduler == nil {
|
||||
return noop, nil
|
||||
}
|
||||
grant := writeclaim.SubagentWriteGrant(ctx)
|
||||
if grant == nil || !grant.Declared().Optimistic {
|
||||
return noop, nil
|
||||
}
|
||||
if _, bound := writer.(pathBoundWriter); bound {
|
||||
return noop, nil // direct legacy callers may have installed their own guard
|
||||
}
|
||||
paths := declared
|
||||
if len(paths) == 0 {
|
||||
resolver, ok := writer.(tool.WritePathResolver)
|
||||
if !ok || !tool.WritesNamedPaths(writer) {
|
||||
return noop, nil
|
||||
}
|
||||
var err error
|
||||
paths, err = resolver.WritePaths(args)
|
||||
if err != nil {
|
||||
return noop, err
|
||||
}
|
||||
}
|
||||
if len(paths) == 0 {
|
||||
return noop, nil
|
||||
}
|
||||
claim, err := parentResolvedWriteReservation(grant.Declared().WorkspaceRoot, paths)
|
||||
if err != nil {
|
||||
return noop, err
|
||||
}
|
||||
return a.svc.writeScheduler.ReserveWrite(claim)
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/base/testenv"
|
||||
"reasonix/internal/runtime/writeclaim"
|
||||
)
|
||||
|
||||
type admissionNamedWriter struct {
|
||||
paths []string
|
||||
err error
|
||||
}
|
||||
|
||||
func (admissionNamedWriter) Name() string { return "model_extra_named_writer" }
|
||||
func (admissionNamedWriter) Description() string { return "Write the supplied paths." }
|
||||
func (admissionNamedWriter) Schema() json.RawMessage { return json.RawMessage(`{}`) }
|
||||
func (admissionNamedWriter) ReadOnly() bool { return false }
|
||||
func (admissionNamedWriter) WritesNamedPaths() bool { return true }
|
||||
func (w admissionNamedWriter) WritePaths(json.RawMessage) ([]string, error) {
|
||||
return w.paths, w.err
|
||||
}
|
||||
func (admissionNamedWriter) Execute(context.Context, json.RawMessage) (string, error) {
|
||||
return "done", nil
|
||||
}
|
||||
|
||||
func TestOptimisticNamedAdmissionUsesTheActorScopeAndReleases(t *testing.T) {
|
||||
root := testenv.TempDir(t)
|
||||
claim, err := writeclaim.OptimisticWorkspaceClaim(root)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := writeclaim.WithSubagentWriteGrant(t.Context(), writeclaim.NewWriteGrant(claim))
|
||||
sched := writeclaim.NewSubagentScheduler(4, 4)
|
||||
a := &Agent{agentConfig: agentConfig{subagentDepth: 1}, svc: agentServices{writeScheduler: sched}}
|
||||
path := filepath.Join(root, "a.txt")
|
||||
writer := admissionNamedWriter{paths: []string{path}}
|
||||
release, err := a.reserveToolWrite(ctx, writer, nil, false, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reservation, err := writeclaim.NormalizeWritePaths(root, []string{path})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if other, err := sched.ReserveWrite(reservation); !errors.Is(err, writeclaim.ErrWritePathBusy) {
|
||||
if other != nil {
|
||||
other()
|
||||
}
|
||||
t.Fatalf("live call was not exclusive: %v", err)
|
||||
}
|
||||
release()
|
||||
other, err := sched.ReserveWrite(reservation)
|
||||
if err != nil {
|
||||
t.Fatalf("reservation survived the call while its actor context remained live: %v", err)
|
||||
}
|
||||
other()
|
||||
if a.writeWorkspaceRoot != "" || a.observeRoot != "" {
|
||||
t.Fatal("claim admission enabled a workspace observation root")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOptimisticNamedAdmissionAcquiresTheWholeSetOrNothing(t *testing.T) {
|
||||
root := testenv.TempDir(t)
|
||||
claim, err := writeclaim.OptimisticWorkspaceClaim(root)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := writeclaim.WithSubagentWriteGrant(t.Context(), writeclaim.NewWriteGrant(claim))
|
||||
sched := writeclaim.NewSubagentScheduler(4, 4)
|
||||
a := &Agent{agentConfig: agentConfig{subagentDepth: 1}, svc: agentServices{writeScheduler: sched}}
|
||||
aPath, bPath := filepath.Join(root, "a.txt"), filepath.Join(root, "b.txt")
|
||||
bClaim, err := writeclaim.NormalizeWritePaths(root, []string{bPath})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
held, err := sched.ReserveWrite(bClaim)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer held()
|
||||
if release, err := a.reserveToolWrite(ctx, admissionNamedWriter{paths: []string{aPath, bPath}}, nil, false, nil); !errors.Is(err, writeclaim.ErrWritePathBusy) {
|
||||
if release != nil {
|
||||
release()
|
||||
}
|
||||
t.Fatalf("overlapping set accepted: %v", err)
|
||||
}
|
||||
aClaim, err := writeclaim.NormalizeWritePaths(root, []string{aPath})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
free, err := sched.ReserveWrite(aClaim)
|
||||
if err != nil {
|
||||
t.Fatalf("refused multi-path call retained its disjoint prefix: %v", err)
|
||||
}
|
||||
free()
|
||||
}
|
||||
@@ -35,6 +35,7 @@ func (t *TaskTool) subagentOptions(ctx context.Context, maxSteps int, pricing *p
|
||||
DeliveryProfile: t.deliveryProfile,
|
||||
Ablation: t.ablation,
|
||||
WorkspaceLease: t.workspaceLease,
|
||||
WriteScheduler: t.scheduler,
|
||||
RecoveryGate: t.recoveryGate,
|
||||
RecoveryAgentID: "subagent",
|
||||
RecoveryTaskID: recoveryTaskID,
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
- Fixed: optimistic task/fleet file writes honor live sibling claims, including dynamically resolved local tools; disjoint writes stay available. Refs #14025, #14133.
|
||||
- 修复:乐观 task/fleet 的文件写入会检查兄弟任务占用,包含动态本地工具;不相交写入仍可并行。关联 #14025、#14133。
|
||||
Reference in new issue
Block a user