rogee/go-sip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bound approved Agent calls by signed duration

Browse Source
This commit is contained in:
rogee
2026-09-29 22:29:02 +08:00
parent c61858fded
commit 428d587f45
5 changed files with 186 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/callflow/capture.go
+9
View File
@@ -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")
}
internal/callflow/capture_cancel_test.go
+44
View File
@@ -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")
}
}
internal/callflow/flow.go
+2 -6
View File
@@ -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 {
internal/rpc/approved_runner.go
+28
View File
@@ -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))
}
internal/rpc/approved_runner_test.go
+103
View File
@@ -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)
}
}