toki
dc9a9a8c59
승인된 execution preset을 Edge 조정 경계와 Node workspace/tool 실행 경계로 연결해 단일 요청 수명주기와 관측 계약을 일관되게 처리한다.
415 lines
14 KiB
Go
415 lines
14 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type channelFakeExecutor struct {
|
|
fn func(context.Context, SingleRequestRequest, SingleRequestController) error
|
|
}
|
|
|
|
func (f *channelFakeExecutor) ExecuteSingleRequest(ctx context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if f.fn != nil {
|
|
return f.fn(ctx, req, ctrl)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func createTestBinding(t *testing.T) *SingleRequestBinding {
|
|
t.Helper()
|
|
binding, err := NewSingleRequestBinding(
|
|
"test-model", "workspace-ref-123",
|
|
SingleRequestStageBinding{Model: "gemini-3.6-flash", Options: map[string]any{"reasoning_effort": "high"}},
|
|
SingleRequestStageBinding{Model: "ornith-fast"},
|
|
SingleRequestStageBinding{Model: "gemini-3.6-flash", Options: map[string]any{"reasoning_effort": "high"}},
|
|
SingleRequestLimits{WallClockMS: 60000, StageTimeoutMS: 10000, MaxToolIterations: 10, MaxOutputBytes: 1048576},
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("NewSingleRequestBinding: %v", err)
|
|
}
|
|
return binding
|
|
}
|
|
|
|
func testEnvelope(requestID string, sequence uint64, stage SingleRequestState) SingleRequestEnvelope {
|
|
return SingleRequestEnvelope{RequestID: requestID, Sequence: sequence, Stage: stage}
|
|
}
|
|
|
|
func submitToFinalizing(req SingleRequestRequest, ctrl SingleRequestController, result *SingleRequestResult) error {
|
|
for sequence, stage := range []SingleRequestState{
|
|
SingleRequestStatePlanning,
|
|
SingleRequestStateWorking,
|
|
SingleRequestStateReviewing,
|
|
SingleRequestStateFinalizing,
|
|
} {
|
|
env := testEnvelope(req.RequestID, uint64(sequence+1), stage)
|
|
if stage == SingleRequestStateFinalizing {
|
|
env.Result = result
|
|
}
|
|
if err := ctrl.SubmitEnvelope(env); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func waitForState(t *testing.T, handle SingleRequestExecution, want SingleRequestState) {
|
|
t.Helper()
|
|
deadline := time.After(2 * time.Second)
|
|
ticker := time.NewTicker(time.Millisecond)
|
|
defer ticker.Stop()
|
|
for {
|
|
if handle.State() == want {
|
|
return
|
|
}
|
|
select {
|
|
case <-deadline:
|
|
t.Fatalf("state=%s, want %s", handle.State(), want)
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func waitForExecution(t *testing.T, handle SingleRequestExecution) (SingleRequestResult, error) {
|
|
t.Helper()
|
|
type outcome struct {
|
|
result SingleRequestResult
|
|
err error
|
|
}
|
|
done := make(chan outcome, 1)
|
|
go func() {
|
|
result, err := handle.Wait()
|
|
done <- outcome{result: result, err: err}
|
|
}()
|
|
select {
|
|
case outcome := <-done:
|
|
return outcome.result, outcome.err
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("Wait did not return")
|
|
return SingleRequestResult{}, nil
|
|
}
|
|
}
|
|
|
|
func startTestExecution(t *testing.T, executor SingleRequestExecutor) SingleRequestExecution {
|
|
t.Helper()
|
|
handle, err := startSingleRequest(context.Background(), executor, SingleRequestRequest{
|
|
RequestID: "request-test",
|
|
Binding: createTestBinding(t),
|
|
Prompt: "complete the private task",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("StartSingleRequest: %v", err)
|
|
}
|
|
return handle
|
|
}
|
|
|
|
func TestSingleRequestExecutorUnavailable(t *testing.T) {
|
|
_, err := (&Service{}).StartSingleRequest(context.Background(), SingleRequestRequest{
|
|
RequestID: "request-unavailable",
|
|
Binding: createTestBinding(t),
|
|
})
|
|
if !errors.Is(err, ErrSingleRequestExecutorUnavailable) {
|
|
t.Fatalf("error=%v, want ErrSingleRequestExecutorUnavailable", err)
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestExecutorCannotMutateAdmission(t *testing.T) {
|
|
result := &SingleRequestResult{Output: "accepted result"}
|
|
executor := &channelFakeExecutor{fn: func(_ context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
req.Binding.Plan.Options["reasoning_effort"] = "low"
|
|
ctrl.Binding().Review.Options["reasoning_effort"] = "low"
|
|
if err := submitToFinalizing(req, ctrl, result); err != nil {
|
|
return err
|
|
}
|
|
result.Output = "executor-mutated result"
|
|
return nil
|
|
}}
|
|
|
|
callerBinding := createTestBinding(t)
|
|
handle, err := startSingleRequest(context.Background(), executor, SingleRequestRequest{
|
|
RequestID: "request-test",
|
|
Binding: callerBinding,
|
|
Prompt: "complete the private task",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("StartSingleRequest: %v", err)
|
|
}
|
|
callerBinding.Plan.Options["reasoning_effort"] = "caller-mutated"
|
|
waitForState(t, handle, SingleRequestStateFinalizing)
|
|
if got := handle.Binding().Plan.Options["reasoning_effort"]; got != "high" {
|
|
t.Fatalf("executor mutated retained binding: %v", got)
|
|
}
|
|
if err := handle.AcknowledgeTerminal(true); err != nil {
|
|
t.Fatalf("AcknowledgeTerminal: %v", err)
|
|
}
|
|
got, err := waitForExecution(t, handle)
|
|
if err != nil || got.Output != "accepted result" {
|
|
t.Fatalf("Wait=(%q, %v), want accepted immutable result", got.Output, err)
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestRejectsExecutorCompletedEnvelope(t *testing.T) {
|
|
executor := &channelFakeExecutor{fn: func(_ context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if err := submitToFinalizing(req, ctrl, &SingleRequestResult{Output: "candidate"}); err != nil {
|
|
return err
|
|
}
|
|
return ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 5, SingleRequestStateCompleted))
|
|
}}
|
|
handle := startTestExecution(t, executor)
|
|
_, err := waitForExecution(t, handle)
|
|
if !errors.Is(err, ErrSingleRequestInvalidState) {
|
|
t.Fatalf("Wait error=%v, want invalid state", err)
|
|
}
|
|
if got := handle.State(); got != SingleRequestStateFailed {
|
|
t.Fatalf("state=%s, want failed", got)
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestExecutorExitFailsClosed(t *testing.T) {
|
|
for name, executor := range map[string]SingleRequestExecutor{
|
|
"no envelopes": &channelFakeExecutor{},
|
|
"mid-stage": &channelFakeExecutor{fn: func(_ context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
return ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 1, SingleRequestStatePlanning))
|
|
}},
|
|
} {
|
|
t.Run(name, func(t *testing.T) {
|
|
handle := startTestExecution(t, executor)
|
|
_, err := waitForExecution(t, handle)
|
|
if !errors.Is(err, ErrSingleRequestFailed) {
|
|
t.Fatalf("Wait error=%v, want ErrSingleRequestFailed", err)
|
|
}
|
|
if got := handle.State(); got != SingleRequestStateFailed {
|
|
t.Fatalf("state=%s, want failed", got)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestEnvelopeOrderingFailsClosed(t *testing.T) {
|
|
tests := map[string]func(SingleRequestRequest, SingleRequestController) error{
|
|
"missing sequence": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
return ctrl.SubmitEnvelope(SingleRequestEnvelope{RequestID: req.RequestID, Stage: SingleRequestStatePlanning})
|
|
},
|
|
"duplicate sequence": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 1, SingleRequestStatePlanning)); err != nil {
|
|
return err
|
|
}
|
|
return ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 1, SingleRequestStateWorking))
|
|
},
|
|
"reordered sequence": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 2, SingleRequestStatePlanning)); err != nil {
|
|
return err
|
|
}
|
|
return ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 1, SingleRequestStateWorking))
|
|
},
|
|
"mismatched saved stage": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 1, SingleRequestStatePlanning)); err != nil {
|
|
return err
|
|
}
|
|
tool := testEnvelope(req.RequestID, 2, SingleRequestStateInternalTool)
|
|
tool.SavedStage = SingleRequestStateWorking
|
|
return ctrl.SubmitEnvelope(tool)
|
|
},
|
|
"duplicate internal tool": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 1, SingleRequestStatePlanning)); err != nil {
|
|
return err
|
|
}
|
|
tool := testEnvelope(req.RequestID, 2, SingleRequestStateInternalTool)
|
|
tool.SavedStage = SingleRequestStatePlanning
|
|
if err := ctrl.SubmitEnvelope(tool); err != nil {
|
|
return err
|
|
}
|
|
tool.Sequence = 3
|
|
return ctrl.SubmitEnvelope(tool)
|
|
},
|
|
}
|
|
for name, submit := range tests {
|
|
t.Run(name, func(t *testing.T) {
|
|
handle := startTestExecution(t, &channelFakeExecutor{fn: func(_ context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
return submit(req, ctrl)
|
|
}})
|
|
_, err := waitForExecution(t, handle)
|
|
if !errors.Is(err, ErrSingleRequestInvalidSequence) && !errors.Is(err, ErrSingleRequestInvalidState) {
|
|
t.Fatalf("Wait error=%v, want sequence or state failure", err)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestFinalCandidateRequired(t *testing.T) {
|
|
tests := map[string]func(SingleRequestRequest, SingleRequestController) error{
|
|
"nil finalizing candidate": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
for sequence, stage := range []SingleRequestState{
|
|
SingleRequestStatePlanning,
|
|
SingleRequestStateWorking,
|
|
SingleRequestStateReviewing,
|
|
SingleRequestStateFinalizing,
|
|
} {
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, uint64(sequence+1), stage)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
},
|
|
"stale earlier-stage candidate": func(req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
env := testEnvelope(req.RequestID, 1, SingleRequestStatePlanning)
|
|
env.Result = &SingleRequestResult{Output: "stale candidate"}
|
|
return ctrl.SubmitEnvelope(env)
|
|
},
|
|
}
|
|
for name, submit := range tests {
|
|
t.Run(name, func(t *testing.T) {
|
|
handle := startTestExecution(t, &channelFakeExecutor{fn: func(_ context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
return submit(req, ctrl)
|
|
}})
|
|
_, err := waitForExecution(t, handle)
|
|
if !errors.Is(err, ErrSingleRequestInvalidState) {
|
|
t.Fatalf("Wait error=%v, want ErrSingleRequestInvalidState", err)
|
|
}
|
|
if got := handle.State(); got != SingleRequestStateFailed {
|
|
t.Fatalf("state=%s, want failed", got)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestAcknowledgementRequiresFinalCandidate(t *testing.T) {
|
|
executor := &channelFakeExecutor{fn: func(ctx context.Context, _ SingleRequestRequest, _ SingleRequestController) error {
|
|
<-ctx.Done()
|
|
return ctx.Err()
|
|
}}
|
|
handle := startTestExecution(t, executor)
|
|
internal := handle.(*singleRequestHandle)
|
|
internal.mu.Lock()
|
|
internal.state = SingleRequestStateFinalizing
|
|
internal.mu.Unlock()
|
|
|
|
err := handle.AcknowledgeTerminal(true)
|
|
if !errors.Is(err, ErrSingleRequestInvalidState) {
|
|
t.Fatalf("AcknowledgeTerminal error=%v, want ErrSingleRequestInvalidState", err)
|
|
}
|
|
if got := handle.State(); got != SingleRequestStateFailed {
|
|
t.Fatalf("state=%s, want failed", got)
|
|
}
|
|
if _, err := waitForExecution(t, handle); !errors.Is(err, ErrSingleRequestInvalidState) {
|
|
t.Fatalf("Wait error=%v, want ErrSingleRequestInvalidState", err)
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestProgressRedactionAndFinalCandidateDelivery(t *testing.T) {
|
|
ready := make(chan struct{})
|
|
release := make(chan struct{})
|
|
executor := &channelFakeExecutor{fn: func(_ context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
planning := testEnvelope(req.RequestID, 1, SingleRequestStatePlanning)
|
|
planning.Message = "raw executor secret"
|
|
if err := ctrl.SubmitEnvelope(planning); err != nil {
|
|
return err
|
|
}
|
|
close(ready)
|
|
<-release
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 2, SingleRequestStateWorking)); err != nil {
|
|
return err
|
|
}
|
|
if err := ctrl.SubmitEnvelope(testEnvelope(req.RequestID, 3, SingleRequestStateReviewing)); err != nil {
|
|
return err
|
|
}
|
|
final := testEnvelope(req.RequestID, 4, SingleRequestStateFinalizing)
|
|
final.Message = "raw executor secret"
|
|
final.Result = &SingleRequestResult{Output: "raw executor result"}
|
|
return ctrl.SubmitEnvelope(final)
|
|
}}
|
|
handle := startTestExecution(t, executor)
|
|
<-ready
|
|
|
|
// Saturate the ordinary lane before the final candidate is emitted.
|
|
internal := handle.(*singleRequestHandle)
|
|
internal.mu.Lock()
|
|
for range 100 {
|
|
internal.notifyProgressLocked(SingleRequestProgress{RequestID: internal.req.RequestID, Message: "filler"}, false)
|
|
}
|
|
internal.mu.Unlock()
|
|
close(release)
|
|
waitForState(t, handle, SingleRequestStateFinalizing)
|
|
|
|
seenFinalizingCandidate := false
|
|
for len(internal.progressCh) > 0 {
|
|
progress := <-internal.progressCh
|
|
if progress.Message == "raw executor secret" || progress.Err != nil {
|
|
t.Fatalf("progress leaked executor-controlled data: %#v", progress)
|
|
}
|
|
if progress.Result == nil {
|
|
if progress.Stage == SingleRequestStateFinalizing {
|
|
t.Fatal("finalizing progress did not include a final candidate")
|
|
}
|
|
continue
|
|
}
|
|
if progress.Stage != SingleRequestStateFinalizing {
|
|
t.Fatalf("non-finalizing progress exposed a result: %#v", progress)
|
|
}
|
|
if progress.Result.Output != "raw executor result" {
|
|
t.Fatalf("finalizing result=%q, want final candidate", progress.Result.Output)
|
|
}
|
|
progress.Result.Output = "surface-mutated result"
|
|
seenFinalizingCandidate = true
|
|
}
|
|
if !seenFinalizingCandidate {
|
|
t.Fatal("finalizing candidate was dropped after ordinary progress saturation")
|
|
}
|
|
if err := handle.AcknowledgeTerminal(true); err != nil {
|
|
t.Fatalf("AcknowledgeTerminal: %v", err)
|
|
}
|
|
result, err := waitForExecution(t, handle)
|
|
if err != nil {
|
|
t.Fatalf("Wait: %v", err)
|
|
}
|
|
if result.Output != "raw executor result" {
|
|
t.Fatalf("Wait result=%q, want immutable final candidate", result.Output)
|
|
}
|
|
}
|
|
|
|
func TestSingleRequestTerminalRaces(t *testing.T) {
|
|
for i := 0; i < 20; i++ {
|
|
release := make(chan struct{})
|
|
executor := &channelFakeExecutor{fn: func(ctx context.Context, req SingleRequestRequest, ctrl SingleRequestController) error {
|
|
if err := submitToFinalizing(req, ctrl, &SingleRequestResult{Output: "candidate"}); err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case <-release:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}}
|
|
handle := startTestExecution(t, executor)
|
|
waitForState(t, handle, SingleRequestStateFinalizing)
|
|
|
|
var wg sync.WaitGroup
|
|
wg.Add(4)
|
|
go func() { defer wg.Done(); _ = handle.AcknowledgeTerminal(true) }()
|
|
go func() { defer wg.Done(); _ = handle.AcknowledgeTerminal(false) }()
|
|
go func() { defer wg.Done(); handle.Cancel() }()
|
|
go func() {
|
|
defer wg.Done()
|
|
_ = handle.SubmitEnvelope(SingleRequestEnvelope{RequestID: "request-test", Sequence: 5, Stage: SingleRequestStateFailed, Err: errors.New("executor failure")})
|
|
}()
|
|
wg.Wait()
|
|
close(release)
|
|
if _, err := waitForExecution(t, handle); err == nil && handle.State() != SingleRequestStateCompleted {
|
|
t.Fatalf("non-completed terminal state must retain an error")
|
|
}
|
|
|
|
terminalCount := 0
|
|
for progress := range handle.Progress() {
|
|
if isTerminalState(progress.Stage) {
|
|
terminalCount++
|
|
}
|
|
}
|
|
if terminalCount != 1 {
|
|
t.Fatalf("terminal progress count=%d, want exactly one", terminalCount)
|
|
}
|
|
}
|
|
}
|