mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
Merge pull request #14684 from SivanCola/codex/ancestor-copy-admission-14025
fix(delegation): reserve ancestor copy admission / 预留祖先副本续跑准入
This commit is contained in:
11 files changed
+729
-7
No files matched your search
@@ -0,0 +1,24 @@
|
||||
[
|
||||
{
|
||||
"package": "reasonix/internal/runtime/delegation",
|
||||
"tests": [
|
||||
"TestAncestorCopyAdmissionCoversUnpublishedRun",
|
||||
"TestAncestorCopyAdmissionCoversUnpublishedRun/continue_then_continue",
|
||||
"TestAncestorCopyAdmissionCoversUnpublishedRun/continue_then_legacy",
|
||||
"TestAncestorCopyAdmissionCoversUnpublishedRun/legacy_then_continue",
|
||||
"TestAncestorCopyAdmissionCoversUnpublishedRun/legacy_then_legacy",
|
||||
"TestAncestorCopyAdmissionAllowsDifferentParents",
|
||||
"TestAncestorCopyAdmissionReleasesAbandonedPreparation",
|
||||
"TestAncestorCopyAdmissionReleasesAfterInvalidCopy",
|
||||
"TestAncestorCopyAdmissionAllowsDifferentSources",
|
||||
"TestAncestorCopyAdmissionUsesNearestSourceIdentity",
|
||||
"TestAncestorCopyAdmissionReleasesAfterBusyCurrentReference",
|
||||
"TestTaskAncestorCopyAdmissionBeforeRunningMetadata",
|
||||
"TestTaskAncestorCopyAdmissionBeforeRunningMetadata/continue_from",
|
||||
"TestTaskAncestorCopyAdmissionBeforeRunningMetadata/fork_from",
|
||||
"TestTaskAncestorCopyAdmissionReleasesUnstartedBackground",
|
||||
"TestTaskAncestorCopyAdmissionReleasesUnstartedBackground/continue_from",
|
||||
"TestTaskAncestorCopyAdmissionReleasesUnstartedBackground/fork_from"
|
||||
]
|
||||
}
|
||||
]
|
||||
@@ -170,6 +170,23 @@ jobs:
|
||||
shell: pwsh
|
||||
run: Get-CimInstance Win32_Processor | Select-Object -ExpandProperty Name
|
||||
|
||||
- name: test (Windows concurrency contracts)
|
||||
if: env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
|
||||
timeout-minutes: 5
|
||||
shell: bash
|
||||
run: |
|
||||
node --test scripts/windows-concurrency-tests.test.mjs
|
||||
node scripts/windows-concurrency-tests.mjs "$RUNNER_TEMP/concurrency-windows.jsonl"
|
||||
|
||||
- name: upload Windows concurrency inventory
|
||||
if: always() && env.RUN_STEPS == 'true' && runner.os == 'Windows' && github.event_name == 'pull_request'
|
||||
uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
|
||||
with:
|
||||
name: concurrency-windows-${{ github.run_attempt }}
|
||||
path: ${{ runner.temp }}/concurrency-windows.jsonl
|
||||
if-no-files-found: error
|
||||
retention-days: 7
|
||||
|
||||
- name: test
|
||||
if: env.RUN_STEPS == 'true' && runner.os != 'Windows'
|
||||
env:
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
---
|
||||
owner: @esengine
|
||||
backup: @SivanCola
|
||||
status: active
|
||||
reviewed: 2026-10-28
|
||||
---
|
||||
|
||||
# Ancestor continuation admission / 祖先续跑准入
|
||||
|
||||
## Behavior / 行为
|
||||
|
||||
1. Within one `SubagentStore`, continuing an ancestor transcript reserves its
|
||||
canonical nearest source and target parent session before searching for a
|
||||
current-session copy. The reservation lasts until the returned run releases.
|
||||
2. A competing continuation for that pair is refused as already in progress,
|
||||
including before the first copy publishes its `running` metadata. It MUST
|
||||
NOT create another copy or make another provider request.
|
||||
3. `continue_from` and the accepted legacy `fork_from` input share this owner.
|
||||
Referring to an older ancestor or its nearest lineage copy uses the same
|
||||
reservation. Different target parents and different sources remain separate.
|
||||
4. Failed preparation and abandoned unpublished runs release their reservation.
|
||||
After a completed run releases, another continuation reuses its persisted
|
||||
copy. Direct current-reference continuation keeps its existing ref lock.
|
||||
|
||||
1. 同一 `SubagentStore` 在查找当前会话副本前,按最近祖先来源和目标父会话预留续跑归属;预留随返回的 run 一起释放。
|
||||
2. 首份副本尚未发布 `running` 元数据时,同一组合的竞争续跑也会被拒绝;不得再创建副本或发起模型请求。
|
||||
3. `continue_from` 与兼容的旧 `fork_from` 输入共用此归属。较早祖先及其最近副本归一到同一来源;不同目标会话、不同来源保持独立。
|
||||
4. 准备失败或放弃未发布的 run 会释放预留。已完成 run 释放后,续跑复用原持久化副本;直接按当前副本引用续跑仍使用原引用锁。
|
||||
|
||||
`errors.Is(err, ErrAncestorCopyBusy)` identifies the refusal through wrapping.
|
||||
TaskTool uses the existing `subagent.settling` answer code and preserves that
|
||||
sentinel; callers do not classify error prose.
|
||||
|
||||
包装后的错误仍可用 `errors.Is(err, ErrAncestorCopyBusy)` 识别;TaskTool 使用既有
|
||||
`subagent.settling` 回答码并保留此标识,不解析错误文案。
|
||||
|
||||
## Compatibility and limits / 兼容与边界
|
||||
|
||||
| Boundary / 边界 | Contract / 合同 |
|
||||
| --- | --- |
|
||||
| Storage / 存储 | No metadata, transcript, status or schema change. / 不改变元数据、转录、状态或 schema。 |
|
||||
| Existing ambiguity / 既有歧义 | Multiple existing copies remain refused; no automatic selection or deletion. / 已有多个副本仍拒绝,不自动选择或删除。 |
|
||||
| Scope / 范围 | One Store instance; existing process-local locking contract. / 仅同一 Store 实例,沿用进程内锁合同。 |
|
||||
| Cross-version writers / 跨版本写入 | No cross-process or old-writer fence; old binaries retain their behavior. / 不提供跨进程或旧写入者隔离。 |
|
||||
| Recovery / 恢复 | No resume of failed/interrupted children, durable reservation or restart recovery. / 不恢复失败或中断子任务,不提供持久化预留与重启恢复。 |
|
||||
| Authority / 权限 | Tool permission, cancellation authority, provider prefix and tool schema are unchanged. / 工具权限、取消权限、模型前缀和工具 schema 不变。 |
|
||||
|
||||
## Validation / 验证
|
||||
|
||||
1. Store regressions hold an unpublished preparation, cover both input routes,
|
||||
nearest-source aliases, independent parents/sources and release after errors.
|
||||
2. TaskTool regressions block the existing parent-destruction callback before
|
||||
`MarkRunning` writes metadata. A competing call is refused; the first call
|
||||
receives the real persisted source transcript and creates the only copy.
|
||||
3. These are local Store/TaskTool tests with a scripted provider, not a live
|
||||
provider, full Boot, browser, crash/restart or physical-process-scope test.
|
||||
4. The Windows workflow requires every named regression and the package terminal
|
||||
pass, with no skips. Local cross-compilation is not native Windows evidence.
|
||||
|
||||
1. Store 回归保留未发布的准备结果,覆盖两种输入、最近来源别名、不同会话和来源,以及错误后的释放。
|
||||
2. TaskTool 回归在 `MarkRunning` 写元数据前阻塞既有父会话销毁检查回调。竞争调用被拒绝,首个调用读取真实持久化来源转录并创建唯一副本。
|
||||
3. 以上是本机 Store/TaskTool 与脚本 provider 验证,不是线上 provider、完整 Boot、浏览器、崩溃重启或物理进程范围验证。
|
||||
4. Windows 工作流要求全部指定测试及包终态通过且零跳过;本机交叉编译不等于 Windows 原生执行。
|
||||
|
||||
A second TaskTool regression requests background execution without a job manager.
|
||||
Preparation succeeds, then normal completion abandons the unstarted run.
|
||||
No test calls `Release` for it; a foreground retry proves admission was freed.
|
||||
|
||||
另一 TaskTool 回归在没有 job manager 时请求后台执行:准备成功后由正常结束路径
|
||||
放弃尚未启动的 run;测试不调用其 `Release`,前台重试证明预留已释放。
|
||||
|
||||
The shared `scripts/windows-concurrency-tests.mjs` step reads additive,
|
||||
package-qualified inventories under `.github/windows-concurrency/`. This suite
|
||||
requires 17 named passes including parents. The broad smoke remains.
|
||||
|
||||
共享脚本读取 `.github/windows-concurrency/` 中按包限定的独立清单;此套件要求含
|
||||
父测试的 17 个指定测试通过,保留 broad smoke。
|
||||
@@ -0,0 +1,63 @@
|
||||
package delegation
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ErrAncestorCopyBusy means this store already holds a continuation of the
|
||||
// same canonical source in the target parent. Callers may retry after release.
|
||||
var ErrAncestorCopyBusy = errors.New("ancestor continuation is already in progress")
|
||||
|
||||
type ancestorCopyKey struct {
|
||||
parent string
|
||||
source string
|
||||
}
|
||||
|
||||
// A copied run has no durable metadata until its caller publishes it. Reserve
|
||||
// the target-parent/source pair before looking for an existing copy, and keep
|
||||
// that reservation for the returned run's ordinary lifetime. The ref lock alone
|
||||
// cannot protect this gap because each preparation would mint a different ref.
|
||||
func (s *SubagentStore) prepareContinueFromAncestor(ref string, spec SubagentSpec) (*SubagentRun, error) {
|
||||
source, err := s.nearestLineageSource(ref, spec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
releaseCopy, err := s.reserveAncestorCopy(ancestorCopyKey{
|
||||
parent: strings.TrimSpace(spec.ParentSession), source: source,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
run, err := s.prepareContinueFromSource(source, spec)
|
||||
if err != nil {
|
||||
releaseCopy()
|
||||
return nil, err
|
||||
}
|
||||
releaseRun := run.release
|
||||
run.release = func() {
|
||||
if releaseRun != nil {
|
||||
releaseRun()
|
||||
}
|
||||
releaseCopy()
|
||||
}
|
||||
return run, nil
|
||||
}
|
||||
|
||||
func (s *SubagentStore) reserveAncestorCopy(key ancestorCopyKey) (func(), error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.copying[key] {
|
||||
return nil, fmt.Errorf("%w: subagent reference %q in parent session %q; retry after it finishes", ErrAncestorCopyBusy, key.source, key.parent)
|
||||
}
|
||||
if s.copying == nil {
|
||||
s.copying = make(map[ancestorCopyKey]bool)
|
||||
}
|
||||
s.copying[key] = true
|
||||
return func() {
|
||||
s.mu.Lock()
|
||||
delete(s.copying, key)
|
||||
s.mu.Unlock()
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
package delegation
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
func ancestorCopyFixture(t *testing.T) (*SubagentStore, string, SubagentSpec) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
s := NewSubagentStore(filepath.Join(dir, "subagents"))
|
||||
spec := testSubagentSpec(t, "review")
|
||||
spec.ParentSession = "root"
|
||||
source, err := s.PrepareFresh(spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer source.Release()
|
||||
source.Session.Add(provider.Message{Role: provider.RoleUser, Content: "original transcript bytes"})
|
||||
if err := s.SaveCompleted(source); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
saveTestBranchMeta(t, dir, "root", "")
|
||||
saveTestBranchMeta(t, dir, "child", "root")
|
||||
saveTestBranchMeta(t, dir, "sibling", "root")
|
||||
spec.ParentSession = "child"
|
||||
return s, source.Ref, spec
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionCoversUnpublishedRun(t *testing.T) {
|
||||
for _, firstLegacy := range []bool{false, true} {
|
||||
for _, secondLegacy := range []bool{false, true} {
|
||||
firstName, secondName := "continue", "continue"
|
||||
if firstLegacy {
|
||||
firstName = "legacy"
|
||||
}
|
||||
if secondLegacy {
|
||||
secondName = "legacy"
|
||||
}
|
||||
t.Run(firstName+"_then_"+secondName, func(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
prepare := func(legacy bool) (*SubagentRun, error) {
|
||||
if legacy {
|
||||
return s.PrepareLegacyForkFrom(source, spec)
|
||||
}
|
||||
return s.PrepareContinue(source, spec)
|
||||
}
|
||||
first, err := prepare(firstLegacy)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer first.Release()
|
||||
if _, err := os.Stat(s.metaPath(first.Ref)); !os.IsNotExist(err) {
|
||||
t.Fatalf("preparation published metadata: %v", err)
|
||||
}
|
||||
second, err := prepare(secondLegacy)
|
||||
if second != nil {
|
||||
second.Release()
|
||||
}
|
||||
if !errors.Is(fmt.Errorf("caller: %w", err), ErrAncestorCopyBusy) {
|
||||
t.Fatalf("second admission = %v; first copy has not been published", err)
|
||||
}
|
||||
if err := s.SaveCompleted(first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first.Release()
|
||||
reused, err := prepare(secondLegacy)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer reused.Release()
|
||||
if reused.Ref != first.Ref {
|
||||
t.Fatalf("continued %s instead of original copy %s", reused.Ref, first.Ref)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionAllowsDifferentParents(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
first, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer first.Release()
|
||||
spec.ParentSession = "sibling"
|
||||
second, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatalf("independent target parent blocked: %v", err)
|
||||
}
|
||||
defer second.Release()
|
||||
if first.Ref == second.Ref || second.Meta.ParentSession != "sibling" {
|
||||
t.Fatal("independent target parent did not get its own copy")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionReleasesAbandonedPreparation(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
first, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first.Release()
|
||||
first.Release()
|
||||
second, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatalf("abandoned preparation retained admission: %v", err)
|
||||
}
|
||||
defer second.Release()
|
||||
if second.Ref == first.Ref {
|
||||
t.Fatal("unpublished preparation was adopted as persisted state")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionReleasesAfterInvalidCopy(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
copy, err := s.PrepareFresh(spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
copy.Meta.ForkedFrom = source
|
||||
if err := s.SaveFailed(copy); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
copy.Release()
|
||||
if run, err := s.PrepareContinue(source, spec); err == nil || !strings.Contains(err.Error(), "failed") {
|
||||
if run != nil {
|
||||
run.Release()
|
||||
}
|
||||
t.Fatalf("failed existing copy admission = %v", err)
|
||||
}
|
||||
// Repair only the fixture, to check that the failed lookup released its
|
||||
// transient reservation. Production does not recover failed transcripts.
|
||||
meta, err := s.LoadMeta(copy.Ref)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
meta.Status = sessionstore.SubagentCompleted
|
||||
if err := s.saveMeta(meta); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatalf("failed preparation retained admission: %v", err)
|
||||
}
|
||||
run.Release()
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionAllowsDifferentSources(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
otherSpec := spec
|
||||
otherSpec.ParentSession = "root"
|
||||
other, err := s.PrepareFresh(otherSpec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.SaveCompleted(other); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
other.Release()
|
||||
first, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer first.Release()
|
||||
second, err := s.PrepareContinue(other.Ref, spec)
|
||||
if err != nil {
|
||||
t.Fatalf("different ancestor source was blocked: %v", err)
|
||||
}
|
||||
second.Release()
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionUsesNearestSourceIdentity(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
child, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.SaveCompleted(child); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
child.Release()
|
||||
saveTestBranchMeta(t, filepath.Dir(s.dir), "grandchild", "child")
|
||||
spec.ParentSession = "grandchild"
|
||||
first, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer first.Release()
|
||||
second, err := s.PrepareContinue(child.Ref, spec)
|
||||
if second != nil {
|
||||
second.Release()
|
||||
}
|
||||
if !errors.Is(err, ErrAncestorCopyBusy) {
|
||||
t.Fatalf("alias of nearest ancestor bypassed admission: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAncestorCopyAdmissionReleasesAfterBusyCurrentReference(t *testing.T) {
|
||||
s, source, spec := ancestorCopyFixture(t)
|
||||
copy, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.SaveCompleted(copy); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
copy.Release()
|
||||
direct, err := s.PrepareContinue(copy.Ref, spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer direct.Release()
|
||||
if run, err := s.PrepareContinue(source, spec); err == nil || !strings.Contains(err.Error(), "already") {
|
||||
if run != nil {
|
||||
run.Release()
|
||||
}
|
||||
t.Fatalf("busy current copy admission = %v", err)
|
||||
}
|
||||
direct.Release()
|
||||
viaAncestor, err := s.PrepareContinue(source, spec)
|
||||
if err != nil {
|
||||
t.Fatalf("busy-ref failure retained ancestor reservation: %v", err)
|
||||
}
|
||||
defer viaAncestor.Release()
|
||||
if viaAncestor.Ref != copy.Ref {
|
||||
t.Fatal("ancestor retry created another copy")
|
||||
}
|
||||
}
|
||||
@@ -85,8 +85,9 @@ type SubagentStore struct {
|
||||
// cleanupBeforeReread is a test seam for deterministic lease interleavings.
|
||||
cleanupBeforeReread func(parentSession, ref string)
|
||||
|
||||
mu sync.Mutex
|
||||
locked map[string]bool
|
||||
mu sync.Mutex
|
||||
locked map[string]bool
|
||||
copying map[ancestorCopyKey]bool
|
||||
// live holds the conversation of every run this process is executing, so a
|
||||
// reader sees a child while it works: its file is written only at the end.
|
||||
live map[string]*liveRun
|
||||
@@ -376,11 +377,7 @@ func (s *SubagentStore) PrepareLegacyForkFrom(ref string, spec SubagentSpec) (*S
|
||||
return s.PrepareContinue(ref, spec)
|
||||
}
|
||||
|
||||
func (s *SubagentStore) prepareContinueFromAncestor(sourceRef string, spec SubagentSpec) (*SubagentRun, error) {
|
||||
sourceRef, err := s.nearestLineageSource(sourceRef, spec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
func (s *SubagentStore) prepareContinueFromSource(sourceRef string, spec SubagentSpec) (*SubagentRun, error) {
|
||||
copies, err := s.compatibleCopiesFromSource(sourceRef, spec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -593,6 +593,9 @@ func (t *TaskTool) runProfileSpec(ctx context.Context, spec ProfileExecSpec, exi
|
||||
agent.AttachReviewReport(subReg, grant)
|
||||
run, err := t.prepareTranscriptRunWithPrompt(withUpstream(ctx, spec.Context.Upstream), subReg, modelRef, effortRef, spec.Context.parentSession(ctx), execution, spec.Context.executionParent(execution), spec.Context.ContinueFrom, spec.Context.ForkFrom, spec.Worker.SystemPrompt, spec.Worker.Kind, spec.Worker.Name)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrAncestorCopyBusy) {
|
||||
return "", fmt.Errorf("%s: %w", steerCode(ErrSubagentSettling), err)
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
exit.run = run
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
package delegation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/contract/tool"
|
||||
"reasonix/internal/runtime/agent"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
)
|
||||
|
||||
func TestTaskAncestorCopyAdmissionBeforeRunningMetadata(t *testing.T) {
|
||||
for _, route := range []string{"continue_from", "fork_from"} {
|
||||
t.Run(route, func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
store := NewSubagentStore(filepath.Join(dir, "subagents"))
|
||||
prov := &mockProvider{name: "sub", chunks: []provider.Chunk{
|
||||
{Type: provider.ChunkText, Text: "persisted source answer"},
|
||||
{Type: provider.ChunkDone},
|
||||
}}
|
||||
task := NewTaskTool(prov, nil, tool.NewRegistry(), 20, 0, 0, 0, 0, "", "sys", nil, 0, "", "", nil).
|
||||
WithTranscripts(store, t.TempDir(), "base-model", "base-effort")
|
||||
rootCtx := agent.WithParentSession(context.Background(), "root")
|
||||
out, err := task.Execute(rootCtx, []byte(`{"prompt":"source task"}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := subagentRefFromOutput(t, out)
|
||||
saveTestBranchMeta(t, dir, "root", "")
|
||||
saveTestBranchMeta(t, dir, "child", "root")
|
||||
|
||||
entered, release := make(chan struct{}), make(chan struct{})
|
||||
var once sync.Once
|
||||
unblock := func() { once.Do(func() { close(release) }) }
|
||||
var probes atomic.Int32
|
||||
// This existing host callback is consulted by MarkRunning before it
|
||||
// publishes metadata, after the real TaskTool prepared its transcript.
|
||||
store.WithDestroyedChecker(func(parent string) bool {
|
||||
if parent == "child" && probes.Add(1) == 1 {
|
||||
close(entered)
|
||||
<-release
|
||||
}
|
||||
return false
|
||||
})
|
||||
childCtx := agent.WithParentSession(context.Background(), "child")
|
||||
type result struct {
|
||||
output string
|
||||
err error
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
go func() {
|
||||
output, err := task.Execute(childCtx, []byte(`{"prompt":"first child","continue_from":"`+source+`"}`))
|
||||
done <- result{output, err}
|
||||
}()
|
||||
defer unblock()
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("first TaskTool call did not reach metadata publication")
|
||||
}
|
||||
_, secondErr := task.Execute(childCtx, []byte(`{"prompt":"competing child","`+route+`":"`+source+`"}`))
|
||||
unblock()
|
||||
first := <-done
|
||||
if first.err != nil {
|
||||
t.Fatalf("first TaskTool: %v", first.err)
|
||||
}
|
||||
if !errors.Is(secondErr, ErrAncestorCopyBusy) || !strings.HasPrefix(secondErr.Error(), "subagent.settling: ") {
|
||||
t.Errorf("second TaskTool admission = %v; want busy refusal", secondErr)
|
||||
}
|
||||
copies, err := sessionstore.ListSubagentsByParent(dir, "child")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(copies) != 1 || copies[0].Ref != subagentRefFromOutput(t, first.output) {
|
||||
t.Errorf("current-parent copies = %d; want only the first admitted run", len(copies))
|
||||
}
|
||||
if len(prov.requests) != 2 {
|
||||
t.Errorf("provider requests = %d; want source and first child only", len(prov.requests))
|
||||
}
|
||||
found := false
|
||||
for _, message := range prov.lastReq.Messages {
|
||||
found = found || message.Content == "persisted source answer"
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("child did not receive the persisted source transcript")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskAncestorCopyAdmissionReleasesUnstartedBackground(t *testing.T) {
|
||||
for _, route := range []string{"continue_from", "fork_from"} {
|
||||
t.Run(route, func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
store := NewSubagentStore(filepath.Join(dir, "subagents"))
|
||||
prov := &mockProvider{name: "sub", chunks: []provider.Chunk{
|
||||
{Type: provider.ChunkText, Text: "persisted source answer"},
|
||||
{Type: provider.ChunkDone},
|
||||
}}
|
||||
task := NewTaskTool(prov, nil, tool.NewRegistry(), 20, 0, 0, 0, 0, "", "sys", nil, 0, "", "", nil).
|
||||
WithTranscripts(store, t.TempDir(), "base-model", "base-effort")
|
||||
root := agent.WithParentSession(context.Background(), "root")
|
||||
out, err := task.Execute(root, []byte(`{"prompt":"source task"}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := subagentRefFromOutput(t, out)
|
||||
saveTestBranchMeta(t, dir, "root", "")
|
||||
saveTestBranchMeta(t, dir, "child", "root")
|
||||
child := agent.WithParentSession(context.Background(), "child")
|
||||
// TaskTool prepares the transcript before looking up the job manager.
|
||||
// With no manager, its normal completion owner abandons that run before
|
||||
// MarkRunning or a provider call. The test never calls run.Release.
|
||||
_, err = task.Execute(child, []byte(`{"prompt":"unstarted child","run_in_background":true,"`+route+`":"`+source+`"}`))
|
||||
if err == nil || err.Error() != "background execution is not available in this context" {
|
||||
t.Fatalf("unstarted background result = %v", err)
|
||||
}
|
||||
if len(prov.requests) != 1 {
|
||||
t.Fatal("abandoned child reached the provider")
|
||||
}
|
||||
copies, err := sessionstore.ListSubagentsByParent(dir, "child")
|
||||
if err != nil || len(copies) != 0 {
|
||||
t.Fatalf("abandoned run published metadata: %v, %v", copies, err)
|
||||
}
|
||||
out, err = task.Execute(child, []byte(`{"prompt":"retry in foreground","`+route+`":"`+source+`"}`))
|
||||
if err != nil {
|
||||
t.Fatalf("normal abandonment retained the reservation: %v", err)
|
||||
}
|
||||
copies, err = sessionstore.ListSubagentsByParent(dir, "child")
|
||||
if err != nil || len(copies) != 1 || copies[0].Ref != subagentRefFromOutput(t, out) || len(prov.requests) != 2 {
|
||||
t.Fatalf("foreground retry: copies=%v error=%v requests=%d", copies, err, len(prov.requests))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
同一会话并发续跑祖先子任务时,不再重复创建尚未发布的转录副本。
|
||||
|
||||
## 修复
|
||||
|
||||
- 在现有 Store 内预留目标会话与祖先来源,覆盖首次复制到发布元数据之间的窗口。关联 #14025。
|
||||
- 竞争续跑返回忙碌提示;不同会话和来源仍独立,已完成副本继续复用,不改变存储格式或工具权限。关联 #14025。
|
||||
- 忙碌拒绝保留可识别的错误标识;尚未启动的后台准备被放弃后,正常退出路径释放预留,允许前台重试。关联 #14025。
|
||||
|
||||
---
|
||||
|
||||
<!-- lang:en -->
|
||||
|
||||
## Fixed
|
||||
|
||||
- Reserve ancestor continuation copies within one Store before metadata publication, preventing competing calls from creating duplicate transcripts. Refs #14025.
|
||||
- Busy continuations are refused; independent parents and sources remain separate. Completed copies are reused without storage-format or permission changes. Refs #14025.
|
||||
- Busy refusals retain a typed error identity; normal exit releases an abandoned, unstarted background preparation so a foreground retry can proceed. Refs #14025.
|
||||
@@ -0,0 +1,81 @@
|
||||
// Focused Windows concurrency contracts live in additive, package-qualified
|
||||
// inventories. New regressions do not need another workflow step or verifier.
|
||||
import assert from 'node:assert/strict';
|
||||
import { closeSync, mkdirSync, openSync, readFileSync, readdirSync } from 'node:fs';
|
||||
import { dirname, join, resolve } from 'node:path';
|
||||
import { spawnSync } from 'node:child_process';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
export function readInventory(directory) {
|
||||
const files = readdirSync(directory).filter(name => name.endsWith('.json')).sort();
|
||||
assert(files.length > 0, 'no Windows concurrency inventories');
|
||||
const required = new Map();
|
||||
for (const file of files) {
|
||||
const groups = JSON.parse(readFileSync(join(directory, file), 'utf8'));
|
||||
assert(Array.isArray(groups) && groups.length > 0, `${file}: empty inventory`);
|
||||
for (const group of groups) {
|
||||
assert(group && /^reasonix\/internal\/(?:[a-z0-9_]+\/)*[a-z0-9_]+$/.test(group.package), `${file}: invalid package`);
|
||||
assert(Array.isArray(group.tests) && group.tests.length > 0, `${file}: empty tests`);
|
||||
const tests = required.get(group.package) ?? new Set();
|
||||
for (const test of group.tests) {
|
||||
assert(typeof test === 'string' && /^Test[A-Za-z0-9_]+(?:\/[^\s]+)?$/.test(test), `${file}: invalid test`);
|
||||
tests.add(test);
|
||||
}
|
||||
required.set(group.package, tests);
|
||||
}
|
||||
}
|
||||
for (const tests of required.values()) {
|
||||
for (const test of tests) assert(tests.has(test.split('/')[0]), `missing parent for ${test}`);
|
||||
}
|
||||
return required;
|
||||
}
|
||||
|
||||
export function verifyInventory(required, raw) {
|
||||
const passed = new Map();
|
||||
const packages = new Set();
|
||||
for (const line of raw.split(/\r?\n/).filter(line => line.trim())) {
|
||||
const event = JSON.parse(line);
|
||||
assert(required.has(event.Package), `unexpected package: ${event.Package}`);
|
||||
assert(!['skip', 'fail'].includes(event.Action), `${event.Package}: ${event.Action}: ${event.Test ?? '(package)'}`);
|
||||
if (event.Action !== 'pass') continue;
|
||||
if (!event.Test) {
|
||||
packages.add(event.Package);
|
||||
continue;
|
||||
}
|
||||
const tests = passed.get(event.Package) ?? new Set();
|
||||
assert(!tests.has(event.Test), `duplicate pass: ${event.Package}/${event.Test}`);
|
||||
tests.add(event.Test);
|
||||
passed.set(event.Package, tests);
|
||||
}
|
||||
let count = 0;
|
||||
for (const [pkg, tests] of required) {
|
||||
assert(packages.has(pkg), `missing package terminal: ${pkg}`);
|
||||
for (const test of tests) {
|
||||
assert(passed.get(pkg)?.has(test), `missing test: ${pkg}/${test}`);
|
||||
count++;
|
||||
}
|
||||
}
|
||||
return { requiredTests: count, packages: required.size };
|
||||
}
|
||||
|
||||
function main() {
|
||||
const root = resolve(dirname(fileURLToPath(import.meta.url)), '..');
|
||||
const required = readInventory(join(root, '.github/windows-concurrency'));
|
||||
const roots = [...new Set([...required.values()].flatMap(tests => [...tests].map(test => test.split('/')[0])))].sort();
|
||||
const output = resolve(process.argv[2] ?? join(process.env.RUNNER_TEMP ?? '.', 'concurrency-windows.jsonl'));
|
||||
mkdirSync(dirname(output), { recursive: true });
|
||||
const fd = openSync(output, 'w');
|
||||
let result;
|
||||
try {
|
||||
result = spawnSync('go', ['test', '-count=1', '-timeout=3m', '-json', '-run', `^(${roots.join('|')})$`, ...required.keys()], {
|
||||
cwd: root, stdio: ['ignore', fd, 'inherit'],
|
||||
});
|
||||
} finally {
|
||||
closeSync(fd);
|
||||
}
|
||||
assert(!result.error, result.error?.message);
|
||||
assert.equal(result.status, 0, `go test failed (${result.signal ?? result.status}); see ${output}`);
|
||||
console.log(JSON.stringify({ ...verifyInventory(required, readFileSync(output, 'utf8')), output }));
|
||||
}
|
||||
|
||||
if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) main();
|
||||
@@ -0,0 +1,64 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { spawnSync } from 'node:child_process';
|
||||
import test from 'node:test';
|
||||
import { readInventory, verifyInventory } from './windows-concurrency-tests.mjs';
|
||||
|
||||
test('long invalid subtest suffix is rejected without repeated slash partitioning', () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'concurrency-inventory-'));
|
||||
try {
|
||||
const inventory = tests => JSON.stringify([{ package: 'reasonix/internal/example', tests }]);
|
||||
const file = join(dir, 'names.json');
|
||||
const valid = ['Test0', 'Test0/a/b', 'Test0/!/', 'Test0//'];
|
||||
writeFileSync(file, inventory(valid));
|
||||
assert.deepEqual([...readInventory(dir).values()][0], new Set(valid));
|
||||
writeFileSync(file, inventory(['Test0', `Test0/${'!/'.repeat(10000)} `]));
|
||||
const script = `import assert from 'node:assert/strict';
|
||||
import { readInventory } from ${JSON.stringify(new URL('./windows-concurrency-tests.mjs', import.meta.url).href)};
|
||||
assert.throws(() => readInventory(process.argv[1]), /invalid test/);`;
|
||||
const result = spawnSync(process.execPath, ['--input-type=module', '-e', script, dir], { timeout: 5000, encoding: 'utf8' });
|
||||
assert.ifError(result.error);
|
||||
assert.equal(result.status, 0, result.stderr);
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test('additive manifests keep package-qualified requirements', () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'concurrency-inventory-'));
|
||||
try {
|
||||
const a = 'reasonix/internal/first', b = 'reasonix/internal/second';
|
||||
writeFileSync(join(dir, 'one.json'), JSON.stringify([{ package: a, tests: ['TestShared', 'TestShared/case'] }]));
|
||||
writeFileSync(join(dir, 'two.json'), JSON.stringify([{ package: b, tests: ['TestShared'] }]));
|
||||
const required = readInventory(dir);
|
||||
const events = [
|
||||
{ Package: a, Action: 'pass', Test: 'TestShared/case' },
|
||||
{ Package: a, Action: 'pass', Test: 'TestShared' },
|
||||
{ Package: a, Action: 'pass' },
|
||||
{ Package: b, Action: 'pass', Test: 'TestShared' },
|
||||
{ Package: b, Action: 'pass' },
|
||||
];
|
||||
const raw = values => values.map(value => JSON.stringify(value)).join('\r\n');
|
||||
assert.deepEqual(verifyInventory(required, raw(events)), { requiredTests: 3, packages: 2 });
|
||||
for (let i = 0; i < events.length; i++) {
|
||||
assert.throws(() => verifyInventory(required, raw(events.filter((_, index) => index !== i))));
|
||||
assert.throws(() => verifyInventory(required, raw(events.map((event, index) => index === i ? { ...event, Action: 'skip' } : event))));
|
||||
}
|
||||
assert.throws(() => verifyInventory(required, raw([...events, { Package: a, Action: 'fail' }])));
|
||||
assert.throws(() => verifyInventory(required, raw([...events, events[0]])));
|
||||
assert.throws(() => verifyInventory(required, '{'));
|
||||
writeFileSync(join(dir, 'two.json'), JSON.stringify([{ package: b, tests: ['TestOrphan/case'] }]));
|
||||
assert.throws(() => readInventory(dir));
|
||||
writeFileSync(join(dir, 'two.json'), JSON.stringify([{ package: '../other', tests: ['TestShared'] }]));
|
||||
assert.throws(() => readInventory(dir));
|
||||
writeFileSync(join(dir, 'two.json'), '[]');
|
||||
assert.throws(() => readInventory(dir));
|
||||
rmSync(join(dir, 'one.json'));
|
||||
rmSync(join(dir, 'two.json'));
|
||||
assert.throws(() => readInventory(dir));
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
Reference in new issue
Block a user