diff --git a/contracts/examples/invalid/mq-result-failure-missing-message.json b/contracts/examples/invalid/mq-result-failure-missing-message.json index cc528b6..a6e3e4a 100644 --- a/contracts/examples/invalid/mq-result-failure-missing-message.json +++ b/contracts/examples/invalid/mq-result-failure-missing-message.json @@ -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":{}}} diff --git a/contracts/examples/invalid/mq-result-invented-identity.json b/contracts/examples/invalid/mq-result-invented-identity.json index cd9d48b..cceed49 100644 --- a/contracts/examples/invalid/mq-result-invented-identity.json +++ b/contracts/examples/invalid/mq-result-invented-identity.json @@ -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":{}}} diff --git a/contracts/examples/invalid/mq-result-invented-recording-state.json b/contracts/examples/invalid/mq-result-invented-recording-state.json index 2657689..5e1abf5 100644 --- a/contracts/examples/invalid/mq-result-invented-recording-state.json +++ b/contracts/examples/invalid/mq-result-invented-recording-state.json @@ -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"}}} diff --git a/contracts/schema b/contracts/schema index 7b4448f..bfa3380 160000 --- a/contracts/schema +++ b/contracts/schema @@ -1 +1 @@ -Subproject commit 7b4448f3586104cde91509a5d5951575c37a8b2c +Subproject commit bfa3380a5e01f4bfaea918383fceed81548252d1 diff --git a/docs/contracts.md b/docs/contracts.md index 3546807..c23b337 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -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 收讫。 | 位置 | 用途 | | --- | --- | diff --git a/docs/evidence/connected-call-duration-local.md b/docs/evidence/connected-call-duration-local.md new file mode 100644 index 0000000..65872a8 --- /dev/null +++ b/docs/evidence/connected-call-duration-local.md @@ -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 收取仍需独立现场验证。 diff --git a/internal/agent/recording_delivery_test.go b/internal/agent/recording_delivery_test.go index 5d7eb8e..9a2b8ed 100644 --- a/internal/agent/recording_delivery_test.go +++ b/internal/agent/recording_delivery_test.go @@ -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 diff --git a/internal/asterisk/answer.go b/internal/asterisk/answer.go index 57408dd..5b33c71 100644 --- a/internal/asterisk/answer.go +++ b/internal/asterisk/answer.go @@ -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 } } } diff --git a/internal/asterisk/answer_test.go b/internal/asterisk/answer_test.go index ee31965..6c17765 100644 --- a/internal/asterisk/answer_test.go +++ b/internal/asterisk/answer_test.go @@ -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) } } diff --git a/internal/asterisk/call.go b/internal/asterisk/call.go index 873fc1b..3b9fc25 100644 --- a/internal/asterisk/call.go +++ b/internal/asterisk/call.go @@ -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()) } diff --git a/internal/asterisk/call_test.go b/internal/asterisk/call_test.go index e0d2bc9..2233025 100644 --- a/internal/asterisk/call_test.go +++ b/internal/asterisk/call_test.go @@ -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") diff --git a/internal/callflow/result_payload.go b/internal/callflow/result_payload.go index 87bb1c5..5e290cc 100644 --- a/internal/callflow/result_payload.go +++ b/internal/callflow/result_payload.go @@ -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{}, }) diff --git a/internal/callflow/result_payload_test.go b/internal/callflow/result_payload_test.go index e5d9011..f90af5f 100644 --- a/internal/callflow/result_payload_test.go +++ b/internal/callflow/result_payload_test.go @@ -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 diff --git a/internal/dispatcher/cleanup_test.go b/internal/dispatcher/cleanup_test.go index 89119ca..52b1a90 100644 --- a/internal/dispatcher/cleanup_test.go +++ b/internal/dispatcher/cleanup_test.go @@ -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) { diff --git a/internal/rpc/approved_recorded_real.go b/internal/rpc/approved_recorded_real.go index 8754d0f..2e875de 100644 --- a/internal/rpc/approved_recorded_real.go +++ b/internal/rpc/approved_recorded_real.go @@ -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 { diff --git a/internal/rpc/approved_recorded_real_test.go b/internal/rpc/approved_recorded_real_test.go index 88d8cd1..c7b05ee 100644 --- a/internal/rpc/approved_recorded_real_test.go +++ b/internal/rpc/approved_recorded_real_test.go @@ -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) } } diff --git a/internal/rpc/recording_server_flow_test.go b/internal/rpc/recording_server_flow_test.go index 805f809..b4b985b 100644 --- a/internal/rpc/recording_server_flow_test.go +++ b/internal/rpc/recording_server_flow_test.go @@ -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{}, diff --git a/internal/store/result_test.go b/internal/store/result_test.go index 84b300f..a91f259 100644 --- a/internal/store/result_test.go +++ b/internal/store/result_test.go @@ -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{},