Report confirmed connected duration separately from call result interval
This commit is contained in:
@@ -1 +1 @@
|
||||
{"event_id":"call-result-example","event_type":"call.execute.result","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1001,"issued_at":"2026-09-21T01:02:00Z","payload":{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:00:00Z","ended_at":"2026-09-21T01:01:00Z","duration_ms":60000,"outcome":"failed","reason_code":null,"transcript":[],"opt_out":false,"recording":{}}}
|
||||
{"event_id":"call-result-example","event_type":"call.execute.result","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1001,"issued_at":"2026-09-21T01:02:00Z","payload":{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:00:00Z","ended_at":"2026-09-21T01:01:00Z","duration_ms":60000,"answered_at":null,"connected_duration_ms":null,"outcome":"failed","reason_code":null,"transcript":[],"opt_out":false,"recording":{}}}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"event_id":"call-result-example","event_type":"call.execute.result","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1001,"issued_at":"2026-09-21T01:02:00Z","payload":{"task_id":"task-asr","call_id":"legacy-call-id","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:00:00Z","ended_at":"2026-09-21T01:01:00Z","duration_ms":60000,"outcome":"failed","reason_code":null,"reason_message":"storage unavailable","transcript":[],"opt_out":false,"recording":{}}}
|
||||
{"event_id":"call-result-example","event_type":"call.execute.result","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1001,"issued_at":"2026-09-21T01:02:00Z","payload":{"task_id":"task-asr","call_id":"legacy-call-id","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:00:00Z","ended_at":"2026-09-21T01:01:00Z","duration_ms":60000,"answered_at":null,"connected_duration_ms":null,"outcome":"failed","reason_code":null,"reason_message":"storage unavailable","transcript":[],"opt_out":false,"recording":{}}}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"event_id":"call-result-example","event_type":"call.execute.result","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1001,"issued_at":"2026-09-21T01:02:00Z","payload":{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:00:00Z","ended_at":"2026-09-21T01:01:00Z","duration_ms":60000,"outcome":"failed","reason_code":null,"reason_message":"storage unavailable","transcript":[],"opt_out":false,"recording":{"status":"unavailable"}}}
|
||||
{"event_id":"call-result-example","event_type":"call.execute.result","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1001,"issued_at":"2026-09-21T01:02:00Z","payload":{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:00:00Z","ended_at":"2026-09-21T01:01:00Z","duration_ms":60000,"answered_at":null,"connected_duration_ms":null,"outcome":"failed","reason_code":null,"reason_message":"storage unavailable","transcript":[],"opt_out":false,"recording":{"status":"unavailable"}}}
|
||||
|
||||
+1
-1
Submodule contracts/schema updated: 7b4448f358...bfa3380a5e
@@ -10,6 +10,7 @@ SaaS 开发人员与本项目共同维护业务规范、Schema、MQ 拓扑和正
|
||||
|
||||
2026-10-10 原 `event_id` 全链路反馈规则已同步至共享提交 `ef3009d`;本项目变更、核验和 SaaS 开发同步说明见 [event_id 生命周期核验](evidence/event-id-lifecycle-20261010.md)。具体消息定义仍只以共享合同为准;此次本地改动未部署,不代表线上反馈已切换。
|
||||
任务控制回执及逐字段拒绝反馈规则统一见共享仓库的 `mq-task-control-receipt.schema.json`、`saas-dispatcher.md` 和 `examples/mq-control-{ack,rejected}.json`;SaaS 须同步采用这些入口,不以历史归档提案或旧的仅 status 回执作为现行结构。
|
||||
接通时长新增 `answered_at` / `connected_duration_ms`,原 `duration_ms` 不变;共享版本及 SaaS 开发对接说明见 [接通通话时长本地核验](evidence/connected-call-duration-local.md)。本地通过不代表真实呼叫或 SaaS 收讫。
|
||||
|
||||
| 位置 | 用途 |
|
||||
| --- | --- |
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
# 接通通话时长:本地开发与 SaaS 对接说明
|
||||
|
||||
共享契约固定版本:`bfa3380a5e01f4bfaea918383fceed81548252d1`(已推送 `sip-contracts/main`)。本项目开发分支:`feat/connected-call-duration`,尚未部署、合入或推送本项目。
|
||||
|
||||
SaaS 应从 `call.execute.result.payload.connected_duration_ms` 读取接通后的时长,单位毫秒。原 `duration_ms` 保留既有含义,不能作为接通时长。`answered_at` 为原始外呼通道首次 `Up` 事件时间;接通时长从该事件计到同一通道的 `ChannelDestroyed` 事件。确认未接通并结束时为 `answered_at=null`、`connected_duration_ms=0`;缺少可核验时间证据时为 `null`,不得把未知当成零或从录音长度推断。接通后的媒体/业务失败也可能报告 `outcome=failed`,SaaS 不应仅凭 outcome 推算时长。字段在所有新结果中均出现。SaaS 开发人员应按上述共享版本调整解析、展示及测试;发布顺序和生产收到消息尚未验证。
|
||||
|
||||
已验证:Schema、正反例与离线来源检查;结果生成、原通道 Up/Destroyed 计时、无证据/未接通边界的单元测试;`make check`、`make release-check-local`、`go test -race ./...` 和 `./scripts/coverage.sh` 通过,业务覆盖率 73.1%。所有核验为本地隔离环境,未拨号、部署或调用真实外部服务;实际 Asterisk 事件时间及 SaaS 收取仍需独立现场验证。
|
||||
@@ -72,7 +72,7 @@ func testRecordingDelivery(stub *recordingDeliveryRPC) *RecordingDelivery {
|
||||
func TestRecordingDeliveryNoRecordingReportsOnlyAfterConfirmedEnd(t *testing.T) {
|
||||
stub := &recordingDeliveryRPC{}
|
||||
delivery := testRecordingDelivery(stub)
|
||||
payload := []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:00Z","ended_at":"2026-09-21T01:30:05Z","duration_ms":5000,"outcome":"no_answer","reason_code":480,"reason_message":"no answer","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`)
|
||||
payload := []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:00Z","ended_at":"2026-09-21T01:30:05Z","duration_ms":5000,"answered_at":null,"connected_duration_ms":0,"outcome":"no_answer","reason_code":480,"reason_message":"no answer","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`)
|
||||
if err := delivery.Complete(context.Background(), CompletedRecording{ResultPayload: payload}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -209,7 +209,7 @@ func testDirectRecordingDelivery(t *testing.T, putStatus int) (*RecordingDeliver
|
||||
}
|
||||
delivery.Recovery = &RecordingRecovery{Root: root, Upload: UploadClient{AllowInsecureHTTP: true}}
|
||||
completed := CompletedRecording{
|
||||
ResultPayload: []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:00Z","ended_at":"2026-09-21T01:30:20Z","duration_ms":20000,"outcome":"answered","reason_code":null,"reason_message":"answered","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`),
|
||||
ResultPayload: []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:00Z","ended_at":"2026-09-21T01:30:20Z","duration_ms":20000,"answered_at":null,"connected_duration_ms":null,"outcome":"answered","reason_code":null,"reason_message":"answered","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`),
|
||||
Expected: true, RecordingID: "recording-mock", UploadID: "upload-mock", WAV: wav, DurationMS: durationMS,
|
||||
}
|
||||
return delivery, stub, completed, &puts, root
|
||||
|
||||
@@ -4,24 +4,26 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/CyCoreSystems/ari/v5"
|
||||
)
|
||||
|
||||
// awaitDialUp only runs after Dial was submitted. Create enters Stasis before
|
||||
// any SIP request is sent, so StasisStart alone can never prove an answer.
|
||||
func awaitDialUp(ctx context.Context, sub ari.Subscription, channelID string) error {
|
||||
func awaitDialUp(ctx context.Context, sub ari.Subscription, channelID string) (time.Time, error) {
|
||||
if ctx == nil || sub == nil || channelID == "" {
|
||||
return errors.New("native answer wait has no channel subscription")
|
||||
return time.Time{}, errors.New("native answer wait has no channel subscription")
|
||||
}
|
||||
stasis, up := false, false
|
||||
var answeredAt time.Time
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
return time.Time{}, ctx.Err()
|
||||
case event, ok := <-sub.Events():
|
||||
if !ok {
|
||||
return errors.New("native answer subscription closed before confirmed answer")
|
||||
return time.Time{}, errors.New("native answer subscription closed before confirmed answer")
|
||||
}
|
||||
if event == nil {
|
||||
continue
|
||||
@@ -33,23 +35,26 @@ func awaitDialUp(ctx context.Context, sub ari.Subscription, channelID string) er
|
||||
}
|
||||
case *ari.ChannelStateChange:
|
||||
if e.Channel.ID == channelID && e.Channel.State == "Up" {
|
||||
if !up {
|
||||
answeredAt = time.Time(e.Timestamp)
|
||||
}
|
||||
up = true
|
||||
}
|
||||
case *ari.ChannelDestroyed:
|
||||
if e.Channel.ID == channelID {
|
||||
return fmt.Errorf("native outbound channel destroyed before verified answer: cause=%d", e.Cause)
|
||||
return time.Time{}, fmt.Errorf("native outbound channel destroyed before verified answer: cause=%d", e.Cause)
|
||||
}
|
||||
case *ari.ChannelHangupRequest:
|
||||
if e.Channel.ID == channelID {
|
||||
return fmt.Errorf("native outbound channel requested hangup before verified answer: cause=%d", e.Cause)
|
||||
return time.Time{}, fmt.Errorf("native outbound channel requested hangup before verified answer: cause=%d", e.Cause)
|
||||
}
|
||||
case *ari.StasisEnd:
|
||||
if e.Channel.ID == channelID {
|
||||
return errors.New("native outbound channel left Stasis before verified answer")
|
||||
return time.Time{}, errors.New("native outbound channel left Stasis before verified answer")
|
||||
}
|
||||
}
|
||||
if stasis && up {
|
||||
return nil
|
||||
return answeredAt, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -39,23 +39,36 @@ func TestAwaitDialUpRequiresStasisAndPostDialUp(t *testing.T) {
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
|
||||
defer cancel()
|
||||
err := awaitDialUp(ctx, answerEvents{events: ch}, id)
|
||||
if (err == nil) != tc.answer {
|
||||
answeredAt, err := awaitDialUp(ctx, answerEvents{events: ch}, id)
|
||||
if (err == nil) != tc.answer || (!tc.answer && !answeredAt.IsZero()) {
|
||||
t.Fatalf("answer proof mismatch: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAwaitDialUpPreservesOriginalUpTimestamp(t *testing.T) {
|
||||
id := "owned-call"
|
||||
answer := time.Date(2026, 10, 10, 12, 0, 0, 123000000, time.UTC)
|
||||
ch := make(chan ari.Event, 3)
|
||||
ch <- &ari.ChannelStateChange{EventData: ari.EventData{Timestamp: ari.DateTime(answer)}, Channel: ari.ChannelData{ID: id, State: "Up"}}
|
||||
ch <- &ari.ChannelStateChange{EventData: ari.EventData{Timestamp: ari.DateTime(answer.Add(time.Second))}, Channel: ari.ChannelData{ID: id, State: "Up"}}
|
||||
ch <- &ari.StasisStart{Channel: ari.ChannelData{ID: id}}
|
||||
got, err := awaitDialUp(context.Background(), answerEvents{events: ch}, id)
|
||||
if err != nil || !got.Equal(answer) {
|
||||
t.Fatalf("native answer timestamp lost: %v %v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAwaitDialUpFailsClosedOnLostEventsOrCancellation(t *testing.T) {
|
||||
ch := make(chan ari.Event)
|
||||
close(ch)
|
||||
if err := awaitDialUp(context.Background(), answerEvents{events: ch}, "execution-1"); err == nil || !strings.Contains(err.Error(), "closed") {
|
||||
if _, err := awaitDialUp(context.Background(), answerEvents{events: ch}, "execution-1"); err == nil || !strings.Contains(err.Error(), "closed") {
|
||||
t.Fatalf("subscription loss was hidden: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
if err := awaitDialUp(ctx, answerEvents{events: make(chan ari.Event)}, "execution-1"); !errors.Is(err, context.Canceled) {
|
||||
if _, err := awaitDialUp(ctx, answerEvents{events: make(chan ari.Event)}, "execution-1"); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("cancellation was hidden: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -95,6 +95,9 @@ type NativeCall struct {
|
||||
Context context.Context
|
||||
cancel context.CancelCauseFunc
|
||||
monitorDone chan struct{}
|
||||
monitorStop chan struct{}
|
||||
answeredAt time.Time
|
||||
endedAt atomic.Pointer[time.Time]
|
||||
destroyed atomic.Bool
|
||||
verifyEnd bool
|
||||
confirmedEnd bool
|
||||
@@ -126,7 +129,7 @@ func dialWithClient(ctx context.Context, client ari.Client, request NativeDial)
|
||||
|
||||
// lineTest omits media/AI resources and uses a distinct ARI application.
|
||||
func dialWithOptions(ctx context.Context, client ari.Client, request NativeDial, lineTest bool) (_ *NativeCall, err error) {
|
||||
call := &NativeCall{client: client, verifyEnd: lineTest}
|
||||
call := &NativeCall{client: client, verifyEnd: true}
|
||||
defer func() {
|
||||
if err != nil {
|
||||
cleanupErr := call.Close()
|
||||
@@ -186,12 +189,15 @@ func dialWithOptions(ctx context.Context, client ari.Client, request NativeDial,
|
||||
}
|
||||
answerCtx, cancel := context.WithTimeout(ctx, request.AnswerTimeout)
|
||||
defer cancel()
|
||||
if err := awaitDialUp(answerCtx, call.subscription, request.ExecutionID); err != nil {
|
||||
answeredAt, err := awaitDialUp(answerCtx, call.subscription, request.ExecutionID)
|
||||
if err != nil {
|
||||
call.verifyEnd = true // Created channel has no ExternalMedia or bridge yet.
|
||||
return nil, &NativeCallFailure{Phase: "answer_wait", Cause: err}
|
||||
}
|
||||
call.answeredAt = answeredAt
|
||||
if lineTest {
|
||||
call.monitorDone = make(chan struct{})
|
||||
call.monitorStop = make(chan struct{})
|
||||
go call.watch(request.ExecutionID)
|
||||
return call, nil
|
||||
}
|
||||
@@ -235,6 +241,7 @@ func dialWithOptions(ctx context.Context, client ari.Client, request NativeDial,
|
||||
return nil, &NativeCallFailure{Phase: "bridge_add_media", Cause: err}
|
||||
}
|
||||
call.monitorDone = make(chan struct{})
|
||||
call.monitorStop = make(chan struct{})
|
||||
go call.watch(request.ExecutionID)
|
||||
return call, nil
|
||||
}
|
||||
@@ -243,7 +250,7 @@ func (c *NativeCall) watch(channelID string) {
|
||||
defer close(c.monitorDone)
|
||||
for {
|
||||
select {
|
||||
case <-c.Context.Done():
|
||||
case <-c.monitorStop:
|
||||
return
|
||||
case event, open := <-c.subscription.Events():
|
||||
if !open || event == nil {
|
||||
@@ -256,6 +263,9 @@ func (c *NativeCall) watch(channelID string) {
|
||||
switch event.GetType() {
|
||||
case "ChannelDestroyed":
|
||||
if ended, ok := event.(*ari.ChannelDestroyed); ok {
|
||||
if stamp := time.Time(ended.Timestamp); !stamp.IsZero() {
|
||||
c.endedAt.Store(&stamp)
|
||||
}
|
||||
c.destroyed.Store(true)
|
||||
c.cancel(fmt.Errorf("native channel ended: cause=%d", ended.Cause))
|
||||
} else {
|
||||
@@ -264,12 +274,33 @@ func (c *NativeCall) watch(channelID string) {
|
||||
return
|
||||
case "ChannelHangupRequest", "StasisEnd":
|
||||
c.cancel(fmt.Errorf("native channel ended: event=%s", event.GetType()))
|
||||
return
|
||||
// A requested hangup is not the native end timestamp. Keep
|
||||
// observing this channel until ChannelDestroyed or Close.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// AnsweredAt is the original outbound channel's Up event timestamp.
|
||||
func (c *NativeCall) AnsweredAt() time.Time {
|
||||
if c == nil {
|
||||
return time.Time{}
|
||||
}
|
||||
return c.answeredAt
|
||||
}
|
||||
|
||||
// ConnectedEndedAt is the original outbound channel's ChannelDestroyed event timestamp.
|
||||
// An ARI 404 proves end but cannot reconstruct the exact time it happened.
|
||||
func (c *NativeCall) ConnectedEndedAt() time.Time {
|
||||
if c == nil {
|
||||
return time.Time{}
|
||||
}
|
||||
if stamp := c.endedAt.Load(); stamp != nil {
|
||||
return *stamp
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
// EndConfirmed reports an observed ChannelDestroyed or post-Hangup ARI 404.
|
||||
// An accepted Hangup request alone is not evidence of termination.
|
||||
func (c *NativeCall) EndConfirmed() bool {
|
||||
@@ -307,9 +338,6 @@ func (c *NativeCall) Close() error {
|
||||
if c.cancel != nil {
|
||||
c.cancel(context.Canceled)
|
||||
}
|
||||
if c.subscription != nil {
|
||||
c.subscription.Cancel()
|
||||
}
|
||||
if c.external != nil {
|
||||
c.closeErr = errors.Join(c.closeErr, c.external.Hangup())
|
||||
}
|
||||
@@ -336,6 +364,20 @@ func (c *NativeCall) Close() error {
|
||||
slog.Error("native outbound end unconfirmed", "execution_id", c.outbound.ID(), "readbacks", reads, "http_status", NativeCallHTTPStatus(err))
|
||||
}
|
||||
}
|
||||
if c.monitorDone != nil && c.confirmedEnd && !c.destroyed.Load() {
|
||||
// The 404 is termination proof, not its timestamp. Allow the
|
||||
// already subscribed destroy event to arrive before stopping the bus.
|
||||
select {
|
||||
case <-c.monitorDone:
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
if c.monitorStop != nil {
|
||||
close(c.monitorStop)
|
||||
}
|
||||
if c.subscription != nil {
|
||||
c.subscription.Cancel()
|
||||
}
|
||||
if c.bridge != nil {
|
||||
c.closeErr = errors.Join(c.closeErr, c.bridge.Delete())
|
||||
}
|
||||
|
||||
@@ -34,22 +34,23 @@ func (b *testBus) Subscribe(_ *ari.Key, _ ...string) ari.Subscription { return b
|
||||
|
||||
type testChannels struct {
|
||||
ari.Channel
|
||||
events chan ari.Event
|
||||
create ari.ChannelCreateRequest
|
||||
createRef *ari.Key
|
||||
media ari.ExternalMediaOptions
|
||||
issued int
|
||||
createErr error
|
||||
dialErr error
|
||||
variableErr error
|
||||
externalErr error
|
||||
dataErr error
|
||||
hangupErr error
|
||||
dataChecks int
|
||||
hungup []string
|
||||
mediaPeerIP string
|
||||
callerID string
|
||||
noAnswer bool
|
||||
events chan ari.Event
|
||||
create ari.ChannelCreateRequest
|
||||
createRef *ari.Key
|
||||
media ari.ExternalMediaOptions
|
||||
issued int
|
||||
createErr error
|
||||
dialErr error
|
||||
variableErr error
|
||||
externalErr error
|
||||
dataErr error
|
||||
hangupErr error
|
||||
dataChecks int
|
||||
hungup []string
|
||||
mediaPeerIP string
|
||||
callerID string
|
||||
noAnswer bool
|
||||
endAfterHangup bool
|
||||
}
|
||||
|
||||
func (c *testChannels) Create(reference *ari.Key, request ari.ChannelCreateRequest) (*ari.ChannelHandle, error) {
|
||||
@@ -116,6 +117,13 @@ func (c *testChannels) Data(key *ari.Key) (*ari.ChannelData, error) {
|
||||
if c.dataErr != nil {
|
||||
return nil, c.dataErr
|
||||
}
|
||||
if c.endAfterHangup {
|
||||
for _, id := range c.hungup {
|
||||
if id == key.ID {
|
||||
return nil, nativeHTTPStatusError{code: 404}
|
||||
}
|
||||
}
|
||||
}
|
||||
return &ari.ChannelData{ID: key.ID}, nil
|
||||
}
|
||||
|
||||
@@ -170,7 +178,7 @@ func TestNativeCallFailureReportsStageAndHTTPStatusWithoutResponseBody(t *testin
|
||||
func TestNativeCallBindsSIPCallIDBeforeOneDialAndActualRTPBridge(t *testing.T) {
|
||||
events := make(chan ari.Event, 2)
|
||||
client := &testARIClient{
|
||||
channels: &testChannels{events: events, mediaPeerIP: "127.0.0.1"},
|
||||
channels: &testChannels{events: events, mediaPeerIP: "127.0.0.1", endAfterHangup: true},
|
||||
bridges: &testBridges{},
|
||||
bus: &testBus{sub: answerEvents{events: events}},
|
||||
}
|
||||
@@ -203,6 +211,24 @@ func TestNativeCallBindsSIPCallIDBeforeOneDialAndActualRTPBridge(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNativeCallReportsOnlyOriginalChannelEndTimestamp(t *testing.T) {
|
||||
ended := time.Date(2026, 10, 10, 12, 0, 20, 3000000, time.UTC)
|
||||
events := make(chan ari.Event, 3)
|
||||
ctx, cancel := context.WithCancelCause(context.Background())
|
||||
call := &NativeCall{Context: ctx, cancel: cancel, subscription: answerEvents{events: events}, monitorStop: make(chan struct{}), monitorDone: make(chan struct{})}
|
||||
go call.watch("exec-1")
|
||||
events <- &ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed", Timestamp: ari.DateTime(ended)}, Channel: ari.ChannelData{ID: "someone-else"}}
|
||||
events <- &ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed", Timestamp: ari.DateTime(ended)}, Channel: ari.ChannelData{ID: "exec-1"}}
|
||||
select {
|
||||
case <-call.monitorDone:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("native end event was not observed")
|
||||
}
|
||||
if !call.ConnectedEndedAt().Equal(ended) || !call.EndConfirmed() {
|
||||
t.Fatalf("lost original channel end timestamp: %v", call.ConnectedEndedAt())
|
||||
}
|
||||
}
|
||||
|
||||
func testBindSIPCall(string, string) error { return nil }
|
||||
|
||||
func TestNativeCallStopsMediaOnMatchingCarrierHangup(t *testing.T) {
|
||||
@@ -255,7 +281,7 @@ func TestNativeCallUnknownBridgeOrMediaCreationCleansKnownIdentities(t *testing.
|
||||
for _, step := range []string{"bridge", "external-media"} {
|
||||
t.Run(step, func(t *testing.T) {
|
||||
events := make(chan ari.Event, 2)
|
||||
channels := &testChannels{events: events, mediaPeerIP: "127.0.0.1"}
|
||||
channels := &testChannels{events: events, mediaPeerIP: "127.0.0.1", endAfterHangup: true}
|
||||
bridges := &testBridges{}
|
||||
if step == "bridge" {
|
||||
bridges.createErr = errors.New("lost bridge response")
|
||||
|
||||
@@ -13,17 +13,20 @@ import (
|
||||
// FinalCallFacts are observed or approved call facts, never defaults inferred
|
||||
// from the Agent SDK. The reason code is nil if no actual SIP code is known.
|
||||
type FinalCallFacts struct {
|
||||
TaskID string
|
||||
Callee string
|
||||
TrunkID string
|
||||
StartedAt time.Time
|
||||
EndedAt time.Time
|
||||
Outcome string
|
||||
ReasonCode *int
|
||||
ReasonMessage string
|
||||
SIPStatusLine string
|
||||
SIPRaw string
|
||||
SIPCaptureError string
|
||||
TaskID string
|
||||
Callee string
|
||||
TrunkID string
|
||||
StartedAt time.Time
|
||||
EndedAt time.Time
|
||||
AnsweredAt *time.Time // original Asterisk channel Up event, not media setup time
|
||||
ConnectedEndedAt *time.Time // original Asterisk ChannelDestroyed event, never report/cleanup time
|
||||
ConfirmedUnanswered bool // proven original channel ended before Up
|
||||
Outcome string
|
||||
ReasonCode *int
|
||||
ReasonMessage string
|
||||
SIPStatusLine string
|
||||
SIPRaw string
|
||||
SIPCaptureError string
|
||||
}
|
||||
|
||||
// FinalResultPayload creates the one current MQ result body. Only final
|
||||
@@ -42,6 +45,15 @@ func FinalResultPayload(facts FinalCallFacts, call Result) ([]byte, error) {
|
||||
if facts.Outcome == "no_answer" && len(call.Turns) != 0 {
|
||||
return nil, errors.New("no-answer call cannot contain completed user turns")
|
||||
}
|
||||
if facts.AnsweredAt != nil && (facts.AnsweredAt.IsZero() || facts.AnsweredAt.After(facts.EndedAt) || facts.ConfirmedUnanswered || facts.Outcome == "no_answer") {
|
||||
return nil, errors.New("final call result has contradictory answer evidence")
|
||||
}
|
||||
if facts.ConnectedEndedAt != nil && (facts.AnsweredAt == nil || facts.ConnectedEndedAt.IsZero() || facts.ConnectedEndedAt.Before(*facts.AnsweredAt) || facts.ConnectedEndedAt.After(facts.EndedAt)) {
|
||||
return nil, errors.New("final call result has contradictory native end evidence")
|
||||
}
|
||||
if facts.ConfirmedUnanswered && len(call.Turns) != 0 {
|
||||
return nil, errors.New("unanswered call cannot contain completed user turns")
|
||||
}
|
||||
if facts.ReasonCode != nil && (*facts.ReasonCode < 100 || *facts.ReasonCode > 699) {
|
||||
return nil, errors.New("final call result has an invalid SIP response code")
|
||||
}
|
||||
@@ -69,6 +81,19 @@ func FinalResultPayload(facts FinalCallFacts, call Result) ([]byte, error) {
|
||||
segments := make([]transcriptSegment, 0, len(call.Turns))
|
||||
optOut := false
|
||||
callDuration := facts.EndedAt.Sub(facts.StartedAt).Milliseconds()
|
||||
var answeredAt *string
|
||||
var connectedDuration *int64
|
||||
if facts.AnsweredAt != nil {
|
||||
formatted := facts.AnsweredAt.UTC().Format(time.RFC3339Nano)
|
||||
answeredAt = &formatted
|
||||
if facts.ConnectedEndedAt != nil {
|
||||
duration := facts.ConnectedEndedAt.Sub(*facts.AnsweredAt).Milliseconds()
|
||||
connectedDuration = &duration
|
||||
}
|
||||
} else if facts.ConfirmedUnanswered || facts.Outcome == "no_answer" {
|
||||
zero := int64(0)
|
||||
connectedDuration = &zero
|
||||
}
|
||||
for i, turn := range call.Turns {
|
||||
window := call.CaptureWindows[i]
|
||||
if window.StartedAt.Before(facts.StartedAt) || window.EndedAt.Before(window.StartedAt) || window.EndedAt.After(facts.EndedAt) {
|
||||
@@ -100,25 +125,27 @@ func FinalResultPayload(facts FinalCallFacts, call Result) ([]byte, error) {
|
||||
captureError = &facts.SIPCaptureError
|
||||
}
|
||||
return json.Marshal(struct {
|
||||
TaskID string `json:"task_id"`
|
||||
Callee string `json:"callee"`
|
||||
TrunkID string `json:"trunk_id"`
|
||||
StartedAt string `json:"started_at"`
|
||||
EndedAt string `json:"ended_at"`
|
||||
DurationMS int64 `json:"duration_ms"`
|
||||
Outcome string `json:"outcome"`
|
||||
ReasonCode *int `json:"reason_code"`
|
||||
ReasonMessage string `json:"reason_message"`
|
||||
StatusLine *string `json:"status_line"`
|
||||
Raw *string `json:"raw"`
|
||||
SIPCaptureError *string `json:"sip_capture_error"`
|
||||
Transcript []transcriptSegment `json:"transcript"`
|
||||
OptOut bool `json:"opt_out"`
|
||||
Recording map[string]any `json:"recording"`
|
||||
TaskID string `json:"task_id"`
|
||||
Callee string `json:"callee"`
|
||||
TrunkID string `json:"trunk_id"`
|
||||
StartedAt string `json:"started_at"`
|
||||
EndedAt string `json:"ended_at"`
|
||||
DurationMS int64 `json:"duration_ms"`
|
||||
AnsweredAt *string `json:"answered_at"`
|
||||
ConnectedDurationMS *int64 `json:"connected_duration_ms"`
|
||||
Outcome string `json:"outcome"`
|
||||
ReasonCode *int `json:"reason_code"`
|
||||
ReasonMessage string `json:"reason_message"`
|
||||
StatusLine *string `json:"status_line"`
|
||||
Raw *string `json:"raw"`
|
||||
SIPCaptureError *string `json:"sip_capture_error"`
|
||||
Transcript []transcriptSegment `json:"transcript"`
|
||||
OptOut bool `json:"opt_out"`
|
||||
Recording map[string]any `json:"recording"`
|
||||
}{
|
||||
TaskID: facts.TaskID, Callee: facts.Callee, TrunkID: facts.TrunkID,
|
||||
StartedAt: facts.StartedAt.UTC().Format(time.RFC3339Nano), EndedAt: facts.EndedAt.UTC().Format(time.RFC3339Nano),
|
||||
DurationMS: callDuration, Outcome: facts.Outcome, ReasonCode: facts.ReasonCode,
|
||||
DurationMS: callDuration, AnsweredAt: answeredAt, ConnectedDurationMS: connectedDuration, Outcome: facts.Outcome, ReasonCode: facts.ReasonCode,
|
||||
ReasonMessage: facts.ReasonMessage, StatusLine: statusLine, Raw: raw, SIPCaptureError: captureError,
|
||||
Transcript: segments, OptOut: optOut, Recording: map[string]any{},
|
||||
})
|
||||
|
||||
@@ -72,6 +72,54 @@ func TestFinalResultPayloadUsesOnlyObservedUserASRAndMediaTimes(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestFinalResultPayloadConnectedDuration(t *testing.T) {
|
||||
facts, _ := approvedResultFixture()
|
||||
answered := facts.StartedAt.Add(400 * time.Millisecond)
|
||||
ended := facts.EndedAt.Add(-100 * time.Millisecond)
|
||||
facts.AnsweredAt, facts.ConnectedEndedAt = &answered, &ended
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
facts FinalCallFacts
|
||||
answer any
|
||||
duration any
|
||||
}{
|
||||
{"answered", facts, answered.Format(time.RFC3339Nano), float64(2500)},
|
||||
{"unknown", func() FinalCallFacts { f := facts; f.AnsweredAt, f.ConnectedEndedAt = nil, nil; return f }(), nil, nil},
|
||||
{"unanswered", func() FinalCallFacts {
|
||||
f := facts
|
||||
f.AnsweredAt, f.ConnectedEndedAt = nil, nil
|
||||
f.Outcome, f.ConfirmedUnanswered = "failed", true
|
||||
return f
|
||||
}(), nil, float64(0)},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
data, err := FinalResultPayload(tc.facts, Result{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got map[string]any
|
||||
if err := json.Unmarshal(data, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got["answered_at"] != tc.answer || got["connected_duration_ms"] != tc.duration || got["duration_ms"] != float64(3000) {
|
||||
t.Fatalf("incorrect duration evidence: %+v", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
for _, change := range []func(*FinalCallFacts){
|
||||
func(f *FinalCallFacts) { f.AnsweredAt = nil },
|
||||
func(f *FinalCallFacts) { bad := f.EndedAt.Add(time.Millisecond); f.AnsweredAt = &bad },
|
||||
func(f *FinalCallFacts) { bad := f.AnsweredAt.Add(-time.Millisecond); f.ConnectedEndedAt = &bad },
|
||||
func(f *FinalCallFacts) { f.ConfirmedUnanswered = true },
|
||||
} {
|
||||
f := facts
|
||||
change(&f)
|
||||
if _, err := FinalResultPayload(f, Result{}); err == nil {
|
||||
t.Fatal("contradictory connected time was accepted")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFinalResultPayloadPreservesOriginalSIPResponse(t *testing.T) {
|
||||
facts, _ := approvedResultFixture()
|
||||
code := 480
|
||||
|
||||
@@ -11,7 +11,7 @@ import (
|
||||
)
|
||||
|
||||
func cleanerResultPayload() []byte {
|
||||
return []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:01Z","ended_at":"2026-09-21T01:30:06Z","duration_ms":5000,"outcome":"no_answer","reason_code":486,"reason_message":"busy","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`)
|
||||
return []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:01Z","ended_at":"2026-09-21T01:30:06Z","duration_ms":5000,"answered_at":null,"connected_duration_ms":0,"outcome":"no_answer","reason_code":486,"reason_message":"busy","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`)
|
||||
}
|
||||
|
||||
func TestConfirmedResultCleansCallAndPreservesSourceIDForReuse(t *testing.T) {
|
||||
|
||||
@@ -3,6 +3,7 @@ package rpc
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
@@ -33,9 +34,11 @@ type ApprovedRecordedRealCall struct {
|
||||
}
|
||||
|
||||
type realMedia struct {
|
||||
ctx context.Context
|
||||
session callflow.MediaSession
|
||||
close func() error
|
||||
ctx context.Context
|
||||
session callflow.MediaSession
|
||||
close func() error
|
||||
answerAt time.Time
|
||||
endAt func() time.Time
|
||||
}
|
||||
|
||||
// Prepare rejects missing signed identity and recovery capacity before the
|
||||
@@ -78,7 +81,7 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con
|
||||
if err != nil {
|
||||
return realMedia{}, err
|
||||
}
|
||||
return realMedia{ctx: call.Context, session: call.Media, close: call.Close}, nil
|
||||
return realMedia{ctx: call.Context, session: call.Media, close: call.Close, answerAt: call.AnsweredAt(), endAt: call.ConnectedEndedAt}, nil
|
||||
}
|
||||
}
|
||||
dial := asterisk.NativeDial{
|
||||
@@ -97,6 +100,20 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con
|
||||
}
|
||||
facts.ReasonCode, facts.SIPStatusLine, facts.SIPRaw = &response.Code, response.StatusLine, response.Raw
|
||||
}
|
||||
addConnectedTimes := func(call realMedia, facts *callflow.FinalCallFacts) {
|
||||
if call.answerAt.IsZero() {
|
||||
slog.Warn("native answer timestamp unavailable", "event_id", approved.SourceEventID)
|
||||
return
|
||||
}
|
||||
facts.AnsweredAt = &call.answerAt
|
||||
if call.endAt != nil {
|
||||
if ended := call.endAt(); !ended.IsZero() {
|
||||
facts.ConnectedEndedAt = &ended
|
||||
return
|
||||
}
|
||||
}
|
||||
slog.Warn("native end timestamp unavailable after confirmed end", "event_id", approved.SourceEventID)
|
||||
}
|
||||
var started atomic.Bool
|
||||
return func(ctx context.Context) error {
|
||||
if !started.CompareAndSwap(false, true) {
|
||||
@@ -120,7 +137,7 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con
|
||||
facts := callflow.FinalCallFacts{
|
||||
TaskID: approved.TaskID,
|
||||
Callee: approved.Callee, TrunkID: approved.SelectedTrunkID,
|
||||
StartedAt: attemptedAt, EndedAt: time.Now().UTC(), Outcome: "failed", ReasonMessage: "call ended before answer",
|
||||
StartedAt: attemptedAt, EndedAt: time.Now().UTC(), ConfirmedUnanswered: true, Outcome: "failed", ReasonMessage: "call ended before answer",
|
||||
}
|
||||
addSIPResponse(ctx, &facts)
|
||||
payload, payloadErr := callflow.FinalResultPayload(facts, callflow.Result{})
|
||||
@@ -147,6 +164,7 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con
|
||||
Callee: approved.Callee, TrunkID: approved.SelectedTrunkID,
|
||||
StartedAt: startedAt, EndedAt: time.Now().UTC(), Outcome: "failed", ReasonMessage: reason,
|
||||
}
|
||||
addConnectedTimes(call, &facts)
|
||||
addSIPResponse(ctx, &facts)
|
||||
payload, err := callflow.FinalResultPayload(facts, callflow.Result{})
|
||||
if err != nil {
|
||||
@@ -185,6 +203,7 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con
|
||||
Callee: approved.Callee, TrunkID: approved.SelectedTrunkID,
|
||||
StartedAt: startedAt, EndedAt: endedAt, Outcome: outcome, ReasonMessage: reason,
|
||||
}
|
||||
addConnectedTimes(call, &facts)
|
||||
addSIPResponse(ctx, &facts)
|
||||
payload, err := callflow.FinalResultPayload(facts, observed)
|
||||
if err != nil {
|
||||
|
||||
@@ -73,6 +73,8 @@ func TestApprovedRecordedRealCallReportsOnlyEndedObservedCall(t *testing.T) {
|
||||
fixture, stub, puts, _, approved := recordedMockFixture(t, nil, time.Second)
|
||||
approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second
|
||||
var originate, hangup int
|
||||
answered := time.Now().Add(-2500 * time.Millisecond).UTC()
|
||||
var nativeEnd time.Time
|
||||
runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(),
|
||||
MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second,
|
||||
Delivery: fixture.Delivery,
|
||||
@@ -81,7 +83,7 @@ func TestApprovedRecordedRealCallReportsOnlyEndedObservedCall(t *testing.T) {
|
||||
if dial.ExecutionID != approved.SourceEventID || dial.DialedCallee != approved.DialedCallee || dial.CallerID != approved.CallerID || dial.TrunkID != approved.SelectedTrunkID || dial.MediaPayloadType != 118 {
|
||||
t.Errorf("signed SIP identity or media profile changed: %+v", dial)
|
||||
}
|
||||
return realMedia{ctx: ctx, session: endedRealMedia{}, close: func() error { hangup++; return nil }}, nil
|
||||
return realMedia{ctx: ctx, session: endedRealMedia{}, close: func() error { hangup++; nativeEnd = time.Now().UTC(); return nil }, answerAt: answered, endAt: func() time.Time { return nativeEnd }}, nil
|
||||
},
|
||||
}
|
||||
err := runner.Run(context.Background(), approved)
|
||||
@@ -89,11 +91,13 @@ func TestApprovedRecordedRealCallReportsOnlyEndedObservedCall(t *testing.T) {
|
||||
t.Fatalf("real call must end and report actual media failure once: err=%v originate=%d hangup=%d calls=%v puts=%d", err, originate, hangup, stub.calls, puts.Load())
|
||||
}
|
||||
var result struct {
|
||||
Outcome string `json:"outcome"`
|
||||
Recording map[string]any `json:"recording"`
|
||||
Outcome string `json:"outcome"`
|
||||
Recording map[string]any `json:"recording"`
|
||||
AnsweredAt string `json:"answered_at"`
|
||||
ConnectedDurationMS *int64 `json:"connected_duration_ms"`
|
||||
}
|
||||
if err := json.Unmarshal(stub.result, &result); err != nil || result.Outcome != "failed" || len(result.Recording) != 0 {
|
||||
t.Fatalf("empty media must not become an answered recording: result=%+v err=%v", result, err)
|
||||
if err := json.Unmarshal(stub.result, &result); err != nil || result.Outcome != "failed" || len(result.Recording) != 0 || result.AnsweredAt != answered.Format(time.RFC3339Nano) || result.ConnectedDurationMS == nil || *result.ConnectedDurationMS < 2500 || *result.ConnectedDurationMS > 3000 {
|
||||
t.Fatalf("ended real call must preserve original connected duration even on media failure: result=%+v err=%v", result, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -114,7 +114,7 @@ func recordingResultPayload(t *testing.T, task configread.Snapshot, grant *agent
|
||||
result := map[string]any{
|
||||
"task_id": "task-asr",
|
||||
"callee": "15003164745", "trunk_id": "trunk-mock",
|
||||
"started_at": "2026-09-21T01:30:01Z", "ended_at": "2026-09-21T01:30:06Z", "duration_ms": 5000,
|
||||
"started_at": "2026-09-21T01:30:01Z", "ended_at": "2026-09-21T01:30:06Z", "duration_ms": 5000, "answered_at": nil, "connected_duration_ms": nil,
|
||||
"outcome": "no_answer", "reason_code": 486, "reason_message": "busy",
|
||||
"status_line": nil, "raw": nil, "sip_capture_error": nil,
|
||||
"transcript": []any{}, "opt_out": false, "recording": map[string]any{},
|
||||
|
||||
@@ -17,7 +17,7 @@ func currentResultPayload(t *testing.T) []byte {
|
||||
payload := map[string]any{
|
||||
"task_id": "task-asr",
|
||||
"callee": "15003164745", "trunk_id": "trunk-mock",
|
||||
"started_at": "2026-09-21T01:30:01Z", "ended_at": "2026-09-21T01:30:06Z", "duration_ms": 5000,
|
||||
"started_at": "2026-09-21T01:30:01Z", "ended_at": "2026-09-21T01:30:06Z", "duration_ms": 5000, "answered_at": nil, "connected_duration_ms": nil,
|
||||
"outcome": "no_answer", "reason_code": 486, "reason_message": "busy",
|
||||
"status_line": nil, "raw": nil, "sip_capture_error": nil,
|
||||
"transcript": []any{}, "opt_out": false, "recording": map[string]any{},
|
||||
|
||||
Reference in New Issue
Block a user