Merge tag 'dev-1051' into dev
Release dev-1051 dev-1051
This commit is contained in:
toki
commit
4b1f08452d
2 changed files with 74 additions and 25 deletions
|
|
@ -101,6 +101,20 @@ func executorPlanBody(goal, verification string) []byte {
|
|||
return successBodyWithThoughtSignature(string(b))
|
||||
}
|
||||
|
||||
func executorPlanInspectionBody(id string) []byte {
|
||||
return workToolBody(id, edgeservice.InternalWorkspaceToolRead, `{"relative_path":"README.md"}`)
|
||||
}
|
||||
|
||||
func executorPlanStageResponse(body, id, goal, verification string) ([]byte, bool) {
|
||||
if !strings.Contains(body, "Inspect the workspace before planning") {
|
||||
return nil, false
|
||||
}
|
||||
if strings.Contains(body, `"role":"tool"`) {
|
||||
return executorPlanBody(goal, verification), true
|
||||
}
|
||||
return executorPlanInspectionBody(id), true
|
||||
}
|
||||
|
||||
func executorWorkBody(completion, verification string) []byte {
|
||||
b, _ := json.Marshal(map[string]any{
|
||||
"item_status": "- P1: completed\n- P2: completed",
|
||||
|
|
@ -130,6 +144,7 @@ func TestSingleRequestExecutorInterface(t *testing.T) {
|
|||
|
||||
func TestSingleRequestExecutorPass(t *testing.T) {
|
||||
responses := [][]byte{
|
||||
executorPlanInspectionBody("plan-pass-1"),
|
||||
executorPlanBody("Execute step 1", "Verify step 1"),
|
||||
workToolBody("work-pass-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`),
|
||||
executorWorkBody("Completed work step 1", "Verified work step 1"),
|
||||
|
|
@ -181,6 +196,7 @@ func TestSingleRequestExecutorPass(t *testing.T) {
|
|||
|
||||
func TestSingleRequestExecutorInspection(t *testing.T) {
|
||||
responses := [][]byte{
|
||||
executorPlanInspectionBody("plan-inspect-1"),
|
||||
executorPlanBody("Plan inspect", "Verify plan"),
|
||||
workToolBody("work-inspect-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`),
|
||||
executorWorkBody("Work done", "Work verified"),
|
||||
|
|
@ -229,6 +245,7 @@ func TestSingleRequestExecutorInspection(t *testing.T) {
|
|||
|
||||
func TestSingleRequestExecutorRepair(t *testing.T) {
|
||||
responses := [][]byte{
|
||||
executorPlanInspectionBody("plan-repair-1"),
|
||||
executorPlanBody("Plan repair", "Verify plan"),
|
||||
workToolBody("work-repair-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`),
|
||||
executorWorkBody("Work initial", "Work initial verify"),
|
||||
|
|
@ -309,8 +326,8 @@ func TestSingleRequestExecutorConcurrentToolIsolation(t *testing.T) {
|
|||
bodyStr := string(reqBody)
|
||||
|
||||
var resp []byte
|
||||
if strings.Contains(bodyStr, "Create a concise plan for the task") {
|
||||
resp = executorPlanBody(fmt.Sprintf("Plan for %s", reqID), fmt.Sprintf("Verify plan for %s", reqID))
|
||||
if planResp, ok := executorPlanStageResponse(bodyStr, "plan-inspect-"+reqID, fmt.Sprintf("Plan for %s", reqID), fmt.Sprintf("Verify plan for %s", reqID)); ok {
|
||||
resp = planResp
|
||||
} else if strings.Contains(bodyStr, "Read the supplied plan") {
|
||||
if !strings.Contains(bodyStr, "typed-result-") {
|
||||
resp = workToolBody("colliding-tool-id", edgeservice.InternalWorkspaceToolRead, fmt.Sprintf(`{"relative_path":"output-%s.txt"}`, reqID))
|
||||
|
|
@ -354,6 +371,12 @@ func TestSingleRequestExecutorConcurrentToolIsolation(t *testing.T) {
|
|||
svc, binding, nodeHarness := newTestServiceHarness(t, executor)
|
||||
|
||||
nodeHarness.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
|
||||
if req.GetStageId() == singleRequestPlanStageID {
|
||||
return &iop.WorkspaceToolResponse{
|
||||
RequestId: req.GetRequestId(), StageId: req.GetStageId(), ToolCallId: req.GetToolCallId(),
|
||||
Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS, Content: []byte("plan-inspection-" + req.GetRequestId()),
|
||||
}
|
||||
}
|
||||
toolArrived <- req.GetRequestId()
|
||||
<-releaseToolResponses
|
||||
return &iop.WorkspaceToolResponse{
|
||||
|
|
@ -441,17 +464,17 @@ func TestSingleRequestExecutorConcurrentToolIsolation(t *testing.T) {
|
|||
}
|
||||
|
||||
reqs := nodeHarness.toolRequestsByRequest[reqID]
|
||||
if len(reqs) != 1 {
|
||||
t.Errorf("request %s tool requests count = %d, want 1", reqID, len(reqs))
|
||||
} else if reqs[0].GetToolCallId() != "colliding-tool-id" {
|
||||
t.Errorf("request %s tool call ID = %q, want colliding-tool-id", reqID, reqs[0].GetToolCallId())
|
||||
if len(reqs) != 2 {
|
||||
t.Errorf("request %s tool requests count = %d, want 2", reqID, len(reqs))
|
||||
} else if reqs[0].GetStageId() != singleRequestPlanStageID || reqs[1].GetToolCallId() != "colliding-tool-id" {
|
||||
t.Errorf("request %s tool requests = %+v, want plan inspection then colliding work call", reqID, reqs)
|
||||
}
|
||||
|
||||
resps := nodeHarness.toolResponsesByRequest[reqID]
|
||||
if len(resps) != 1 {
|
||||
t.Errorf("request %s tool responses count = %d, want 1", reqID, len(resps))
|
||||
if len(resps) != 2 {
|
||||
t.Errorf("request %s tool responses count = %d, want 2", reqID, len(resps))
|
||||
} else {
|
||||
gotResult := string(resps[0].GetContent())
|
||||
gotResult := string(resps[1].GetContent())
|
||||
wantResult := fmt.Sprintf("typed-result-%s", reqID)
|
||||
if gotResult != wantResult {
|
||||
t.Errorf("request %s result = %q, want %q", reqID, gotResult, wantResult)
|
||||
|
|
@ -621,11 +644,16 @@ func TestSingleRequestExecutorRequestBudgetOwnership(t *testing.T) {
|
|||
|
||||
func TestSingleRequestExecutorStageFailures(t *testing.T) {
|
||||
t.Run("PlanFailure", func(t *testing.T) {
|
||||
var callCount atomic.Int32
|
||||
mockSvc := &mockService{
|
||||
submit: func(_ context.Context, _ edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
|
||||
body := successBody("invalid plan json")
|
||||
if callCount.Add(1) == 1 {
|
||||
body = executorPlanInspectionBody("plan-failure-inspect")
|
||||
}
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel,
|
||||
Tunnel: &mockTunnel{frames: framesFor(successBody("invalid plan json"))},
|
||||
Tunnel: &mockTunnel{frames: framesFor(body)},
|
||||
DispatchInfo: matchingDispatch(),
|
||||
}, nil
|
||||
},
|
||||
|
|
@ -652,6 +680,11 @@ func TestSingleRequestExecutorStageFailures(t *testing.T) {
|
|||
submit: func(_ context.Context, _ edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
|
||||
count := callCount.Add(1)
|
||||
if count == 1 {
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(executorPlanInspectionBody("work-failure-plan-inspect"))}, DispatchInfo: matchingDispatch(),
|
||||
}, nil
|
||||
}
|
||||
if count == 2 {
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel,
|
||||
Tunnel: &mockTunnel{frames: framesFor(executorPlanBody("Step 1", "Verify 1"))},
|
||||
|
|
@ -687,20 +720,25 @@ func TestSingleRequestExecutorStageFailures(t *testing.T) {
|
|||
submit: func(_ context.Context, _ edgeservice.ProviderPoolDispatchRequest) (*edgeservice.ProviderPoolDispatchResult, error) {
|
||||
count := callCount.Add(1)
|
||||
if count == 1 {
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel, Tunnel: &mockTunnel{frames: framesFor(executorPlanInspectionBody("review-failure-plan-inspect"))}, DispatchInfo: matchingDispatch(),
|
||||
}, nil
|
||||
}
|
||||
if count == 2 {
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel,
|
||||
Tunnel: &mockTunnel{frames: framesFor(executorPlanBody("Step 1", "Verify 1"))},
|
||||
DispatchInfo: matchingDispatch(),
|
||||
}, nil
|
||||
}
|
||||
if count == 2 {
|
||||
if count == 3 {
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel,
|
||||
Tunnel: &mockTunnel{frames: framesFor(workToolBody("review-failure-work-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`))},
|
||||
DispatchInfo: matchingDispatch(),
|
||||
}, nil
|
||||
}
|
||||
if count == 3 {
|
||||
if count == 4 {
|
||||
return &edgeservice.ProviderPoolDispatchResult{
|
||||
Path: edgeservice.ProviderPoolPathTunnel,
|
||||
Tunnel: &mockTunnel{frames: framesFor(executorWorkBody("Work done", "Work verified"))},
|
||||
|
|
@ -733,6 +771,7 @@ func TestSingleRequestExecutorStageFailures(t *testing.T) {
|
|||
|
||||
func TestSingleRequestExecutorFinalOutputProvenance(t *testing.T) {
|
||||
responses := [][]byte{
|
||||
executorPlanInspectionBody("plan-provenance-1"),
|
||||
executorPlanBody("Plan step", "Plan verify"),
|
||||
workToolBody("work-provenance-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`),
|
||||
executorWorkBody("UNAPPROVED WORK CANDIDATE OUTPUT", "Work verified"),
|
||||
|
|
@ -778,8 +817,8 @@ func TestSingleRequestExecutorTerminalWaiterCleanup(t *testing.T) {
|
|||
bodyStr := string(reqBody)
|
||||
|
||||
var resp []byte
|
||||
if strings.Contains(bodyStr, "Create a concise plan for the task") {
|
||||
resp = executorPlanBody("Plan step", "Verify plan")
|
||||
if planResp, ok := executorPlanStageResponse(bodyStr, "plan-clean-success", "Plan step", "Verify plan"); ok {
|
||||
resp = planResp
|
||||
} else if strings.Contains(bodyStr, "Read the supplied plan") {
|
||||
if !strings.Contains(bodyStr, "colliding-tool-id") {
|
||||
resp = workToolBody("colliding-tool-id", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`)
|
||||
|
|
@ -832,8 +871,8 @@ func TestSingleRequestExecutorTerminalWaiterCleanup(t *testing.T) {
|
|||
bodyStr := string(reqBody)
|
||||
|
||||
var resp []byte
|
||||
if strings.Contains(bodyStr, "Create a concise plan for the task") {
|
||||
resp = executorPlanBody("Plan step", "Verify plan")
|
||||
if planResp, ok := executorPlanStageResponse(bodyStr, "plan-clean-failure", "Plan step", "Verify plan"); ok {
|
||||
resp = planResp
|
||||
} else if strings.Contains(bodyStr, "Read the supplied plan") {
|
||||
if !strings.Contains(bodyStr, "colliding-tool-id") {
|
||||
resp = workToolBody("colliding-tool-id", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`)
|
||||
|
|
@ -894,8 +933,8 @@ func TestSingleRequestExecutorTerminalWaiterCleanup(t *testing.T) {
|
|||
}
|
||||
|
||||
var resp []byte
|
||||
if strings.Contains(bodyStr, "Create a concise plan for the task") {
|
||||
resp = executorPlanBody("Plan step for "+reqID, "Verify plan")
|
||||
if planResp, ok := executorPlanStageResponse(bodyStr, "plan-clean-"+reqID, "Plan step for "+reqID, "Verify plan"); ok {
|
||||
resp = planResp
|
||||
} else if strings.Contains(bodyStr, "Read the supplied plan") {
|
||||
if !strings.Contains(bodyStr, "colliding-tool-id") {
|
||||
resp = workToolBody("colliding-tool-id", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"output.txt"}`)
|
||||
|
|
@ -919,7 +958,7 @@ func TestSingleRequestExecutorTerminalWaiterCleanup(t *testing.T) {
|
|||
|
||||
var once sync.Once
|
||||
nodeHarness.toolResponder = func(req *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
|
||||
if req.GetRequestId() == "req-cancel-peer" {
|
||||
if req.GetRequestId() == "req-cancel-peer" && req.GetStageId() == singleRequestWorkStageID {
|
||||
once.Do(func() {
|
||||
cancelWaiterRegistered <- struct{}{}
|
||||
})
|
||||
|
|
|
|||
|
|
@ -308,25 +308,27 @@ func TestSingleRequestQualityGateBudgetAndMalformedCallStopComposite(t *testing.
|
|||
{
|
||||
name: "iteration budget",
|
||||
responses: [][]byte{
|
||||
executorPlanInspectionBody("quality-budget-plan-inspect"),
|
||||
executorPlanBody("Read bounded inputs", "Stop at the bound"),
|
||||
workToolBody("budget-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"first.txt"}`),
|
||||
workToolBody("budget-2", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"second.txt"}`),
|
||||
},
|
||||
maxIterations: 1,
|
||||
want: edgeservice.SingleRequestTerminalDisposition{Kind: edgeservice.SingleRequestTerminalError, ErrorClass: edgeservice.SingleRequestTerminalErrorBudget},
|
||||
wantProviders: 3,
|
||||
wantTools: 1,
|
||||
wantProviders: 4,
|
||||
wantTools: 2,
|
||||
},
|
||||
{
|
||||
name: "malformed workspace call",
|
||||
responses: [][]byte{
|
||||
executorPlanInspectionBody("quality-malformed-plan-inspect"),
|
||||
executorPlanBody("Reject an invalid path", "No tool effect"),
|
||||
workToolBody("malformed-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"../private"}`),
|
||||
},
|
||||
maxIterations: 4,
|
||||
want: edgeservice.SingleRequestTerminalDisposition{Kind: edgeservice.SingleRequestTerminalError, ErrorClass: edgeservice.SingleRequestTerminalErrorMalformed},
|
||||
wantProviders: 2,
|
||||
wantTools: 0,
|
||||
wantProviders: 3,
|
||||
wantTools: 1,
|
||||
},
|
||||
}
|
||||
|
||||
|
|
@ -368,6 +370,7 @@ func TestSingleRequestQualityGateBudgetAndMalformedCallStopComposite(t *testing.
|
|||
|
||||
func TestSingleRequestQualityGateRepetitionStopsBeforeLaterDispatch(t *testing.T) {
|
||||
responses := [][]byte{
|
||||
executorPlanInspectionBody("quality-repeat-plan-inspect"),
|
||||
executorPlanBody("Read once", "Verify once"),
|
||||
workToolBody("repeat-1", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
|
||||
workToolBody("repeat-2", edgeservice.InternalWorkspaceToolRead, `{"relative_path":"result.txt"}`),
|
||||
|
|
@ -399,7 +402,7 @@ func TestSingleRequestQualityGateRepetitionStopsBeforeLaterDispatch(t *testing.T
|
|||
if !errors.Is(waitErr, edgeservice.ErrSingleRequestFailed) || terminal != want {
|
||||
t.Fatalf("Wait=%v terminal=%+v, want repetition", waitErr, terminal)
|
||||
}
|
||||
if providerCalls.Load() != 3 || node.toolCount.Load() != 2 || node.cleanupCount.Load() != 1 || executor.bridge.pendingCount() != 0 || terminalCount != 1 {
|
||||
if providerCalls.Load() != 4 || node.toolCount.Load() != 3 || node.cleanupCount.Load() != 1 || executor.bridge.pendingCount() != 0 || terminalCount != 1 {
|
||||
t.Fatalf("provider=%d tool=%d cleanup=%d pending=%d terminals=%d", providerCalls.Load(), node.toolCount.Load(), node.cleanupCount.Load(), executor.bridge.pendingCount(), terminalCount)
|
||||
}
|
||||
}
|
||||
|
|
@ -447,6 +450,7 @@ func TestSingleRequestQualityGateAdmitsOnlyRepairableToolResults(t *testing.T) {
|
|||
|
||||
func TestSingleRequestQualityGateToolTimeoutStopsBeforeContinuation(t *testing.T) {
|
||||
responses := [][]byte{
|
||||
executorPlanInspectionBody("quality-timeout-plan-inspect"),
|
||||
executorPlanBody("Run command", "Verify command"),
|
||||
workToolBody("timeout-1", edgeservice.InternalWorkspaceToolCommand, `{"command_id":"verify"}`),
|
||||
}
|
||||
|
|
@ -461,6 +465,12 @@ func TestSingleRequestQualityGateToolTimeoutStopsBeforeContinuation(t *testing.T
|
|||
executor := NewSingleRequestExecutor(mockSvc)
|
||||
service, binding, node := newTestServiceHarness(t, executor)
|
||||
node.toolResponder = func(request *iop.WorkspaceToolRequest) *iop.WorkspaceToolResponse {
|
||||
if request.GetStageId() == singleRequestPlanStageID {
|
||||
return &iop.WorkspaceToolResponse{
|
||||
RequestId: request.GetRequestId(), StageId: request.GetStageId(), ToolCallId: request.GetToolCallId(),
|
||||
Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS,
|
||||
}
|
||||
}
|
||||
return &iop.WorkspaceToolResponse{
|
||||
RequestId: request.GetRequestId(), StageId: request.GetStageId(), ToolCallId: request.GetToolCallId(),
|
||||
Status: iop.WorkspaceStatus_WORKSPACE_STATUS_TIMEOUT, ErrorCode: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_TIMEOUT,
|
||||
|
|
@ -484,7 +494,7 @@ func TestSingleRequestQualityGateToolTimeoutStopsBeforeContinuation(t *testing.T
|
|||
if !errors.Is(waitErr, edgeservice.ErrSingleRequestInternalToolFailed) || terminal != want {
|
||||
t.Fatalf("Wait=%v terminal=%+v, want timeout", waitErr, terminal)
|
||||
}
|
||||
if providerCalls.Load() != 2 || node.toolCount.Load() != 1 || node.cleanupCount.Load() != 1 || executor.bridge.pendingCount() != 0 || terminalCount != 1 {
|
||||
if providerCalls.Load() != 3 || node.toolCount.Load() != 2 || node.cleanupCount.Load() != 1 || executor.bridge.pendingCount() != 0 || terminalCount != 1 {
|
||||
t.Fatalf("provider=%d tool=%d cleanup=%d pending=%d terminals=%d", providerCalls.Load(), node.toolCount.Load(), node.cleanupCount.Load(), executor.bridge.pendingCount(), terminalCount)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue