Bind approved caller profile and require explicit media pipeline

docs/evidence/saas-dispatcher-implementation.md

@@ -52,7 +52,7 @@
 - `ExecuteApproved` 只在隔离 Mock 中签发：携带任务原始 JSON、引用的 provider 明文凭据、SIP revision、选定路由/主叫/原始被叫、独立的拨号期限及最大通话时限；任务＋provider 原始字节＋SIP revision 计算绑定摘要。Agent 对照已激活的 D 会话身份、绑定摘要、SDK 能力及**实际观察到的** SIP 加载结果；发出 Mock 指令前仅持久保存摘要和未知占用，失败/重复/重启不自动重拨，文件中不留明文凭据。当前 `LoadedSIP` 为可注入的隔离 Mock 观察器，**不等于真实 Asterisk 已加载核验**。
 - ASR-only 不启动 LLM/TTS；full-AI 使用冻结的 model、显式 temperature=0/max_tokens、TTS voice/format/speech_rate 与 provider 原值凭据。真实 ASR SDK WebSocket 发出的音频输入/识别配置、LLM 与 TTS SDK 向隔离 HTTP Mock 发出的实际参数均已捕获核对；已配置的开场白由 TTS 单次合成，失败不自动重播且未完成时不进入对话轮次；空开场白不调用 TTS。关键词仅匹配用户侧最终 ASR，失败/未知挂断不自动重试。`ApprovedOriginator` 经生成的 Unary gRPC Stub 交付完整快照，SIP 全量 revision 不吻合即拒绝。
 - 对话控制隔离验证：共享 `callflow` 不再根据转写或回复内的硬编码词推断拒联，仅处理明确的关键词/拒联事实；ASR-only 不播报开场或 TTS 回复。配置的首语音/静默期限、整通话期限与最大轮数交给媒体控制器；回复按 Unicode 字符数分片，火山 TTS 的缓存音频块数超限明确失败而不重试；不支持的插话配置在准入前拒绝。该控制器尚未接入新主 CLI，不能宣称真实通话媒体已验证。
-- Agent 隔离媒体入口 `rpc.RunApprovedCall` 仅接收签发的执行快照、媒体会话和挂断动作：签发通话期限与 AI 总期限均约束整通电话；ASR-only 不合成开场，full-AI 开场完成才采集语音。父级期限不会被“未检测到语音”掩盖，空媒体 Mock 等到上下文真正结束。当前媒体只处理 16 kHz PCM16，SDK 虽接受、但媒体不能正确处理的 24 kHz ASR/TTS 配置在 Dispatcher 与 Agent 绑定时明确拒绝，不静默播放或识别错速音频。该入口尚未连接主 CLI，也未用于真实拨号。
+- Agent 隔离媒体入口 `rpc.RunApprovedCall` 仅接收签发的执行快照、媒体会话、挂断动作和显式每通话 AI pipeline；缺失或 typed nil 适配器在读媒体前拒绝，无隐式 SDK/Mock 回退。测试分别由已绑定快照构建 SDK pipeline 与注入隔离 ASR 测试桩。签发通话期限与 AI 总期限均约束整通电话；ASR-only 不合成开场，full-AI 开场完成才采集语音。父级期限不会被“未检测到语音”掩盖，空媒体 Mock 等到上下文真正结束。当前媒体只处理 16 kHz PCM16，SDK 虽接受、但媒体不能正确处理的 24 kHz ASR/TTS 配置在 Dispatcher 与 Agent 绑定时明确拒绝，不静默播放或识别错速音频。该入口尚未连接主 CLI，也未用于真实拨号。
 - 验证：`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-current-contracts.sh`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-proto.sh`、`go test ./... -count=1`、`go test -race ./internal/ai ./internal/callflow ./internal/configread ./internal/dispatcher ./internal/rpc -count=1`、`go vet ./...`、`go build ./...`、`git diff --check` 均通过。
 - **后续边界：** 录音直传/OSS 失败恢复及最终结果属 P06；新主 CLI 接线、真实媒体完整联动及旧路径清理属 P07；隔离端到端和 A01–A12 属 P08。真实 Agent/Asterisk、SaaS、MQ、AI 供应商联调未开展，不能由 Mock 结果代签。
 
@@ -67,7 +67,7 @@
 - Dispatcher 录音事实 Unary RPC 隔离服务：`RequestRecordingUpload`、`ReportCallEnded`、`ReportCallResult` 均要求已配置本 D、核验 mTLS 指纹及当前 Agent 会话、数字租户和已保留的执行；复用官方 SDK 仅对原始录音签发固定 15 分钟授权，显式重申请仍用相同 bucket/object_key。结束事实可先于外呼响应而持久化原回执；录音结果核对 D 已存目标和 Agent 报告的成功 PUT，再与唯一结果 outbox 同事务提交。Mock 覆盖会话/租户拒绝、同资产重申请、上传前结果拒绝、坏 JSON、已上传与无录音结果及重复回报。隔离测试还通过本地双向 TLS 的 gRPC 实际传输：Agent `RecordingClient` 每次读取并克隆当前会话元数据，经受控 D 客户端领取授权、上报结束和唯一结果；未配置客户端明确拒绝。其分层测试不包含 OSS PUT；下述本地隔离链路另验证直传，主入口批准执行的媒体录音仍未接线。
 - 内存录音隔离组件：`RecordingSession` 仅复制共享通话流程实际读到和成功发送的 16-kHz PCM16，`EncodeMonoWAV` 直接在内存生成有界单声道 WAV；空音频、奇数字节、超过上限及未成功发送的音频都不能伪造成可上传录音。单元与 race 测试未产生业务文件。批准执行入口尚未接入该组件，且 Mock 中观测到的帧不等于真实 Asterisk 通话的全量媒体验收。
 - Agent 录音交付隔离组件：`RecordingDelivery` 先确认结束，再依照录音是否实际生成分别上报唯一空录音结果或请求原授权并直传内存 WAV；录音生成失败保留通话真实结果、空录音对象及明确原因，不虚构上传事实。隔离测试通过本地 HTTP PUT 和假 Dispatcher RPC 覆盖成功无业务文件、OSS 明确失败后私有文件保存、恢复写入失败、未知 PUT 隔离、重启重领原目标、上传已确认后只重发原结果。再次调用不会隐式重新 PUT；正常已确认上传但尚未被 D 持久收讫的跨进程间隙仍受 K16 边界约束。此处未连接主入口批准执行媒体或 MQ；下述隔离链路另测本地真正的 D gRPC/SQLite。
-- 最终结果隔离组件：`FinalResultPayload` 只从批准的任务身份、确认时间及实际收到的用户媒体窗口生成当前唯一结果；最终 ASR 文本与字面关键词拒联写入完整转写，未播放的助手候选内容不冒充转写，未知 SIP 响应码保留 `null`，无录音保留 `{}`。单元测试直接核验当前 MQ Schema、字段缺失/乱序时窗与无应答负例；开场白发送失败不得计入已播放音频。此组件尚未接入主入口，也未证明真实 RTP 转写时间精度。
+- 最终结果隔离组件：`ApprovedExecution` 已保留获批任务的 `caller_profile_id`；`FinalResultPayload` 只从批准的任务身份、确认时间及实际收到的用户媒体窗口生成当前唯一结果；最终 ASR 文本与字面关键词拒联写入完整转写，未播放的助手候选内容不冒充转写，未知 SIP 响应码保留 `null`，无录音保留 `{}`。单元测试直接核验当前 MQ Schema、字段缺失/乱序时窗与无应答负例；开场白发送失败不得计入已播放音频。此组件尚未接入主入口，也未证明真实 RTP 转写时间精度。
 - 本地隔离链路：共享批准通话流程 `ExecuteApproved` 让显式 ASR 测试桩按 ASR-only 限制消费实际读出的合成 PCM 媒体帧；`RecordingSession` 生成内存 WAV，`FinalResultPayload` 仅用最终模拟识别及实测采集时窗构造结果。Agent 经双向 TLS gRPC 向 Dispatcher 确认结束、领取原始资产签名授权，再对本地 HTTPS OSS Mock 单次 PUT；Dispatcher 将原上传事实与唯一最终结果 outbox 同事务提交。测试确认一次 PUT、一次结果、无业务文件，使用官方 SDK 生成的路径；ASR 测试桩及 OSS Mock 不验证真实供应商协议或签名。这不是主入口执行、RabbitMQ 投递或外部验收。
 - 已验证：`go test ./... -count=1`、`go test -race ./... -count=1`、`go vet ./...`、`go build ./...`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-current-contracts.sh`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-proto.sh`、`git diff --check`。尚未完成批准执行媒体录音到 Agent 交付组件的主入口接线、主入口 Agent↔Dispatcher 实际会话与录音执行接线（隔离 mTLS gRPC 已测）、真实执行时上传事实/最终结果交付及 MQ/端到端验收，不能宣称 P06 通过。
 

internal/rpc/approved_execution.go

@@ -26,6 +26,7 @@ type ApprovedExecution struct {
 	DispatcherID    string
 	TenantID        int64
 	TaskID          string
+	CallerProfileID string
 	SourceEventID   string
 	CallID          string
 	SelectedTrunkID string
@@ -130,7 +131,7 @@ func (s *Server) ExecuteApproved(ctx context.Context, req *agentpb.ExecuteApprov
 	}
 	approved := ApprovedExecution{
 		DispatcherID: dispatcherID, TenantID: req.TenantId, TaskID: req.TaskId,
-		SourceEventID: req.SourceEventId, CallID: req.CallId,
+		CallerProfileID: task.CallerProfileID, SourceEventID: req.SourceEventId, CallID: req.CallId,
 		SelectedTrunkID: req.SelectedTrunkId, CallerID: req.CallerId,
 		Callee: req.Callee, DialedCallee: req.DialedCallee, SIPRevision: req.SipRevision,
 		RingTimeout:     time.Duration(req.RingTimeoutMs) * time.Millisecond,

internal/rpc/approved_runner.go

@@ -3,6 +3,7 @@ package rpc
 import (
 	"context"
 	"errors"
+	"reflect"
 
 	"git.ipao.vip/rogee/go-sip/internal/ai"
 	"git.ipao.vip/rogee/go-sip/internal/callflow"
@@ -11,18 +12,22 @@ import (
 // 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")
+func missingApprovedAdapter(value any) bool {
+	if value == nil {
+		return true
+	}
+	v := reflect.ValueOf(value)
+	return v.Kind() == reflect.Pointer && v.IsNil()
+}
+
+func RunApprovedCall(ctx context.Context, approved ApprovedExecution, session callflow.MediaSession, hangup func(context.Context) error, pipeline callflow.ApprovedPipeline) (callflow.Result, error) {
+	if missingApprovedAdapter(session) || hangup == nil || missingApprovedAdapter(pipeline) || approved.MaxCallDuration <= 0 {
+		return callflow.Result{}, errors.New("approved Agent media, hangup, pipeline 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))
+	return callflow.ExecuteApproved(ctx, session, ai.Mode(approved.AI.Mode), pipeline, callflow.ApprovedCapture(approved.AI))
 }

internal/rpc/approved_runner_test.go

@@ -1,6 +1,7 @@
 package rpc
 
 import (
+	"bytes"
 	"context"
 	"encoding/base64"
 	"encoding/json"
@@ -14,6 +15,7 @@ import (
 	"time"
 
 	"git.ipao.vip/rogee/go-sip/internal/ai"
+	"git.ipao.vip/rogee/go-sip/internal/callflow"
 	"git.ipao.vip/rogee/go-sip/internal/configread"
 	"git.ipao.vip/rogee/go-sip/internal/media"
 )
@@ -52,7 +54,12 @@ func TestApprovedCallRunnerASROnlyHonorsSignedTimeout(t *testing.T) {
 		t.Fatal(err)
 	}
 	mediaSession := &blockedApprovedMedia{}
-	_, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: bound, MaxCallDuration: 35 * time.Millisecond}, mediaSession, func(context.Context) error { return nil })
+	hangup := func(context.Context) error { return nil }
+	call, err := ai.NewCurrentCall(bound, hangup)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: bound, MaxCallDuration: 35 * time.Millisecond}, mediaSession, hangup, call)
 	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)
 	}
@@ -96,8 +103,45 @@ func TestApprovedCallRunnerSynthesizesOpeningBeforeMediaCapture(t *testing.T) {
 		t.Fatal(err)
 	}
 	mediaSession := &blockedApprovedMedia{}
-	_, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: bound, MaxCallDuration: 100 * time.Millisecond}, mediaSession, func(context.Context) error { return nil })
+	hangup := func(context.Context) error { return nil }
+	call, err := ai.NewCurrentCall(bound, hangup)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: bound, MaxCallDuration: 100 * time.Millisecond}, mediaSession, hangup, call)
 	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)
 	}
 }
+
+func TestApprovedCallRunnerUsesExplicitIsolatedPipelineAndObservedMedia(t *testing.T) {
+	mediaSession, err := callflow.NewRecordingSession(callflow.NewMemorySession(bytes.Repeat([]byte{1, 0}, 1600)), 4096)
+	if err != nil {
+		t.Fatal(err)
+	}
+	approved := ApprovedExecution{
+		AI:              ai.CurrentBound{Mode: "asr_only", Conversation: ai.CurrentConversation{SilenceTimeout: 100 * time.Millisecond, MaxDuration: time.Second, MaxTurns: 1}},
+		MaxCallDuration: time.Second,
+	}
+	result, err := RunApprovedCall(context.Background(), approved, mediaSession, func(context.Context) error { return nil }, isolatedApprovedASR{})
+	if err != nil || len(result.Turns) != 1 || len(result.CaptureWindows) != 1 || len(result.OutboundTurns) != 0 || result.Turns[0].Transcript != "隔离 Mock 最终识别文本" {
+		t.Fatalf("explicit isolated pipeline did not honor approved ASR-only media: turns=%d windows=%d outbound=%d err=%v", len(result.Turns), len(result.CaptureWindows), len(result.OutboundTurns), err)
+	}
+	if wav, durationMS, err := mediaSession.WAV(); err != nil || len(wav) != 3244 || durationMS != 100 {
+		t.Fatalf("approved media did not produce a bounded in-memory recording: size=%d duration=%d err=%v", len(wav), durationMS, err)
+	}
+}
+
+func TestApprovedCallRunnerRejectsTypedNilPipelineBeforeMedia(t *testing.T) {
+	var missing *ai.CurrentCall
+	session := callflow.NewMemorySession(bytes.Repeat([]byte{1, 0}, 1600))
+	_, err := RunApprovedCall(context.Background(), ApprovedExecution{AI: ai.CurrentBound{Mode: "asr_only"}, MaxCallDuration: time.Second}, session, func(context.Context) error { return nil }, missing)
+	if err == nil || session.Stats().ReceivedPackets != 0 {
+		t.Fatalf("missing pipeline was allowed to consume call media: err=%v stats=%+v", err, session.Stats())
+	}
+	var missingMedia *callflow.MemorySession
+	_, err = RunApprovedCall(context.Background(), ApprovedExecution{AI: ai.CurrentBound{Mode: "asr_only"}, MaxCallDuration: time.Second}, missingMedia, func(context.Context) error { return nil }, isolatedApprovedASR{})
+	if err == nil {
+		t.Fatal("typed-nil media session was admitted")
+	}
+}

internal/rpc/approved_snapshot_test.go

@@ -21,7 +21,7 @@ func TestApprovedAgentReceivesFrozenASROnlySnapshot(t *testing.T) {
 	s := activatedApprovedServer(t, now, filepath.Join(t.TempDir(), "session.json"), req.DispatcherId, 1,
 		func(_ context.Context, call ApprovedExecution) error {
 			attempts++
-			if call.DispatcherID != req.DispatcherId || call.TenantID != req.TenantId || call.TaskID != req.TaskId || call.SIPRevision != 8 || call.SelectedTrunkID != req.SelectedTrunkId {
+			if call.DispatcherID != req.DispatcherId || call.TenantID != req.TenantId || call.TaskID != req.TaskId || call.SIPRevision != 8 || call.SelectedTrunkID != req.SelectedTrunkId || call.CallerProfileID != "caller-profile-mock" {
 				t.Fatal("Agent did not receive Dispatcher-authorized call binding")
 			}
 			if call.AI.Mode != "asr_only" || call.AI.LLM != nil || call.AI.TTS != nil || call.AI.ASR.Provider.Credential != "example-only-not-a-real-secret" {