diff --git a/internal/callflow/capture.go b/internal/callflow/capture.go index 74013cf..35f158d 100644 --- a/internal/callflow/capture.go +++ b/internal/callflow/capture.go @@ -38,6 +38,9 @@ func captureTurn(ctx context.Context, session MediaSession, cfg CaptureConfig) ( var frames []byte for { + if err := ctx.Err(); err != nil { + return nil, err + } now := time.Now() if !started && !now.Before(firstDeadline) { return nil, errors.New("no speech detected before capture timeout") @@ -58,6 +61,9 @@ func captureTurn(ctx context.Context, session MediaSession, cfg CaptureConfig) ( readCtx, cancel := context.WithDeadline(ctx, readDeadline) payload, err := session.ReadPayload(readCtx) cancel() + if err := ctx.Err(); err != nil { + return nil, err + } if err != nil { if errors.Is(err, context.DeadlineExceeded) { if started && cfg.EndSilence > 0 && !lastVoice.IsZero() && !time.Now().Before(lastVoice.Add(cfg.EndSilence)) { @@ -90,6 +96,9 @@ func captureTurn(ctx context.Context, session MediaSession, cfg CaptureConfig) ( break } } + if err := ctx.Err(); err != nil { + return nil, err + } if len(frames) == 0 { return nil, errors.New("captured audio is empty") } diff --git a/internal/callflow/capture_cancel_test.go b/internal/callflow/capture_cancel_test.go new file mode 100644 index 0000000..8a7d92e --- /dev/null +++ b/internal/callflow/capture_cancel_test.go @@ -0,0 +1,44 @@ +package callflow + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestCaptureTurnPropagatesParentDeadline(t *testing.T) { + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Millisecond)) + defer cancel() + _, err := captureTurn(ctx, NewMemorySession(nil), CaptureConfig{ + FirstSpeechTimeout: 20 * time.Millisecond, + MaxDuration: 20 * time.Millisecond, + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("a signed call deadline must take precedence over capture timeout: %v", err) + } +} + +func TestMemorySessionWaitsUntilContextEnds(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := NewMemorySession(nil).ReadPayload(ctx) + done <- err + }() + select { + case err := <-done: + t.Fatalf("no inbound media is not a context deadline: %v", err) + case <-time.After(5 * time.Millisecond): + } + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("canceled media read returned %v", err) + } + case <-time.After(time.Second): + t.Fatal("canceled media read did not return") + } +} diff --git a/internal/callflow/flow.go b/internal/callflow/flow.go index c77a206..f2d7658 100644 --- a/internal/callflow/flow.go +++ b/internal/callflow/flow.go @@ -146,12 +146,8 @@ func (m *MemorySession) ReadPayload(ctx context.Context) ([]byte, error) { m.stats.ReceivedBytes += uint64(len(payload)) return append([]byte(nil), payload...), nil } - select { - case <-ctx.Done(): - return nil, ctx.Err() - default: - return nil, context.DeadlineExceeded - } + <-ctx.Done() + return nil, ctx.Err() } func (m *MemorySession) SendPCM16(ctx context.Context, pcm []byte, _ int) error { diff --git a/internal/rpc/approved_runner.go b/internal/rpc/approved_runner.go new file mode 100644 index 0000000..6c75edc --- /dev/null +++ b/internal/rpc/approved_runner.go @@ -0,0 +1,28 @@ +package rpc + +import ( + "context" + "errors" + + "git.ipao.vip/rogee/go-sip/internal/ai" + "git.ipao.vip/rogee/go-sip/internal/callflow" +) + +// RunApprovedCall is the isolated Agent media entry point. Both the +// Dispatcher-signed call limit and the immutable AI conversation limit apply; +// neither missing media nor a missing hangup adapter can silently proceed. +func RunApprovedCall(ctx context.Context, approved ApprovedExecution, session callflow.MediaSession, hangup func(context.Context) error) (callflow.Result, error) { + if session == nil || hangup == nil || approved.MaxCallDuration <= 0 { + return callflow.Result{}, errors.New("approved Agent media, hangup and signed call duration are required") + } + if approved.AI.Mode != string(ai.ModeFullAI) && approved.AI.Mode != string(ai.ModeASROnly) { + return callflow.Result{}, errors.New("approved AI mode is unsupported") + } + call, err := ai.NewCurrentCall(approved.AI, hangup) + if err != nil { + return callflow.Result{}, err + } + ctx, cancel := context.WithTimeout(ctx, approved.MaxCallDuration) + defer cancel() + return callflow.ExecuteApproved(ctx, session, ai.Mode(approved.AI.Mode), call, callflow.ApprovedCapture(approved.AI)) +} diff --git a/internal/rpc/approved_runner_test.go b/internal/rpc/approved_runner_test.go new file mode 100644 index 0000000..189f3c1 --- /dev/null +++ b/internal/rpc/approved_runner_test.go @@ -0,0 +1,103 @@ +package rpc + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "git.ipao.vip/rogee/go-sip/internal/ai" + "git.ipao.vip/rogee/go-sip/internal/configread" + "git.ipao.vip/rogee/go-sip/internal/media" +) + +type blockedApprovedMedia struct { + sent int + rate int +} + +func (*blockedApprovedMedia) ReadPayload(ctx context.Context) ([]byte, error) { + <-ctx.Done() + return nil, ctx.Err() +} +func (m *blockedApprovedMedia) SendPCM16(ctx context.Context, pcm []byte, rate int) error { + if err := ctx.Err(); err != nil { + return err + } + m.sent++ + m.rate = rate + return nil +} +func (*blockedApprovedMedia) Stats() media.RTPStats { return media.RTPStats{} } + +func TestApprovedCallRunnerASROnlyHonorsSignedTimeout(t *testing.T) { + req := approvedTestRequest(t, time.Date(2026, 9, 20, 10, 0, 0, 0, time.UTC)) + var task configread.CurrentTask + if err := json.Unmarshal(req.TaskConfigJson, &task); err != nil { + t.Fatal(err) + } + var providers map[string]configread.CurrentProvider + if err := json.Unmarshal(req.ProvidersJson, &providers); err != nil { + t.Fatal(err) + } + bound, err := ai.BindCurrent(task, providers) + if err != nil { + t.Fatal(err) + } + mediaSession := &blockedApprovedMedia{} + _, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: bound, MaxCallDuration: 35 * time.Millisecond}, mediaSession, func(context.Context) error { return nil }) + if err == nil || !strings.Contains(err.Error(), "deadline") || mediaSession.sent != 0 { + t.Fatalf("ASR-only call must timeout without opening or outbound TTS: sent=%d err=%v", mediaSession.sent, err) + } +} + +func TestApprovedCallRunnerSynthesizesOpeningBeforeMediaCapture(t *testing.T) { + raw, err := os.ReadFile("../../contracts/local/examples/config-read-task-full.json") + if err != nil { + t.Fatal(err) + } + var task configread.CurrentTask + if err := json.Unmarshal(raw, &task); err != nil { + t.Fatal(err) + } + req := approvedTestRequest(t, time.Date(2026, 9, 20, 10, 0, 0, 0, time.UTC)) + var providers map[string]configread.CurrentProvider + if err := json.Unmarshal(req.ProvidersJson, &providers); err != nil { + t.Fatal(err) + } + var sdkCalls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sdkCalls.Add(1) + var payload struct { + Params struct { + Text string `json:"text"` + } `json:"req_params"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil || payload.Params.Text != "Example greeting" { + http.Error(w, "opening text changed", http.StatusBadRequest) + return + } + _, _ = fmt.Fprintf(w, `{"code":0,"data":%q}`+"\n", base64.StdEncoding.EncodeToString([]byte{1, 0, 2, 0})) + _, _ = fmt.Fprintln(w, `{"code":20000000,"message":"ok","data":null}`) + })) + defer server.Close() + tts := providers["tts-example"] + tts.Endpoint = server.URL + providers[tts.ProviderRef] = tts + bound, err := ai.BindCurrent(task, providers) + if err != nil { + t.Fatal(err) + } + mediaSession := &blockedApprovedMedia{} + _, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: bound, MaxCallDuration: 100 * time.Millisecond}, mediaSession, func(context.Context) error { return nil }) + if err == nil || !strings.Contains(err.Error(), "deadline") || sdkCalls.Load() != 1 || mediaSession.sent != 1 || mediaSession.rate != 16000 { + t.Fatalf("approved opening must be played once before timed capture: requests=%d sent=%d rate=%d err=%v", sdkCalls.Load(), mediaSession.sent, mediaSession.rate, err) + } +}