From bbd111254d7d37686d83596c7d7856658c25baa4 Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Tue, 10 Mar 2026 21:07:42 +0900 Subject: [PATCH 1/7] =?UTF-8?q?#280=20=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...0_fix-session-refresh-reinvite-fallback.md | 364 +++++++++ virtual-voicebot-backend/src/main.rs | 17 +- .../src/protocol/session/handlers/mod.rs | 140 +++- .../src/protocol/session/types.rs | 9 +- .../src/protocol/sip/core.rs | 737 +++++++++++++++++- .../src/shared/ports/sip.rs | 11 +- 6 files changed, 1227 insertions(+), 51 deletions(-) create mode 100644 virtual-voicebot-backend/docs/steering/STEER-280_fix-session-refresh-reinvite-fallback.md diff --git a/virtual-voicebot-backend/docs/steering/STEER-280_fix-session-refresh-reinvite-fallback.md b/virtual-voicebot-backend/docs/steering/STEER-280_fix-session-refresh-reinvite-fallback.md new file mode 100644 index 0000000..8f10dc5 --- /dev/null +++ b/virtual-voicebot-backend/docs/steering/STEER-280_fix-session-refresh-reinvite-fallback.md @@ -0,0 +1,364 @@ +# STEER-280: refresher=uas 時の session refresh を Allow ヘッダに従い re-INVITE へフォールバック + +--- + +## 1. メタ情報 + +| 項目 | 値 | +|------|-----| +| ID | STEER-280 | +| タイトル | refresher=uas 時の session refresh を Allow ヘッダに従い re-INVITE へフォールバック | +| ステータス | Approved | +| 関連Issue | #280 | +| 優先度 | P0 | +| 作成日 | 2026-03-10 | + +--- + +## 2. ストーリー（Why） + +### 2.1 背景 + +着信呼で `refresher=uas`（Voicebot が session refresh 送信責任者）として OK DSP を返す場合、一定時間経過後に Voicebot は session refresh を送信する。 +現在の実装は送信メソッドを `UPDATE` に固定しているため、相手の `Allow` ヘッダに `UPDATE` が含まれない場合でも `UPDATE` を送信してしまい、RFC 4028 準拠の `re-INVITE` フォールバックが行われない。 + +**再現条件**: +1. 着信で `refresher=uas` を含む `Session-Expires` を返す（200 OK DSP） +2. 相手の `INVITE` の `Allow` ヘッダに `UPDATE` が含まれない + +**期待動作**: Session-Expires の 80% 経過後に `re-INVITE` を送信 +**実動作**: Session-Expires の 80% 経過後に `UPDATE` を送信（相手が対応していない場合にも） + +### 2.2 目的 + +RFC 4028 §8 の規定「refresher は相手が UPDATE をサポートしていない場合は re-INVITE を使用しなければならない」を満たし、UPDATE 未サポート相手との相互接続性を確保する。 + +### 2.3 ユーザーストーリー + +``` +As a Voicebot オペレーター +I want to UPDATE 未サポートの SIP 端末・PBX との通話が session timer によって強制切断されないこと +So that 長時間通話が正常に継続できる +``` + +受入条件: +- [ ] `Allow: UPDATE` が無い相手への session refresh は `re-INVITE` で送信される +- [ ] `Allow: UPDATE` がある相手への session refresh は従来通り `UPDATE` で送信される +- [ ] `Allow` ヘッダ自体が無い相手への session refresh は `re-INVITE` で送信される（安全側） +- [ ] `re-INVITE` の `2xx` 受信時に ACK が返送される +- [ ] `re-INVITE` の `2xx` 受信時に Session-Expires タイマがリセットされる +- [ ] `re-INVITE` が 408 / 481 の場合は BYE で通話終了する（transaction timeout は本 issue スコープ外） +- [ ] その他の非 `2xx` 応答では即時 BYE せず、session expiration は延長しない（`SessionTimerFired` 発火後 `AppSessionTimeout` 経由で終話） + +--- + +## 3. 段取り（Who / When） + +### 3.1 起票 + +| 項目 | 値 | +|------|-----| +| 起票者 | @MasanoriSuda | +| 起票日 | 2026-03-10 | +| 起票理由 | UPDATE 未サポート端末と接続時に session timer 起因の意図せぬ切断が発生するバグ | + +### 3.2 仕様作成 + +| 項目 | 値 | +|------|-----| +| 作成者 | Claude Sonnet 4.6 | +| 作成日 | 2026-03-10 | +| 指示者 | @MasanoriSuda | +| 指示内容 | "Issue #280 のバグ修正ステアリングファイルの作成（Codex 調査結果を踏まえて）" | + +### 3.3 レビュー + +| # | レビュアー | 日付 | 結果 | コメント | +|---|-----------|------|------|---------| +| 1 | Codex 5.4 | 2026-03-10 | ok | ok | + +### 3.4 承認 + +| 項目 | 値 | +|------|-----| +| 承認者 |@MasanoriSuda | +| 承認日 |2026-03-10 | +| 承認コメント | lgtm | + +### 3.5 実装 + +| 項目 | 値 | +|------|-----| +| 実装者 | Codex | +| 実装日 | 2026-03-10 | +| 指示者 | @MasanoriSuda | +| 指示内容 | "本ステアリング §5 の差分仕様に従い最小差分で実装すること" | +| コードレビュー | 2026-03-10 ローカル品質ゲート通過（`cargo fmt --all -- --check` / `cargo clippy --all-targets --all-features -- -D warnings` / `cargo test --all --all-features`） | + +### 3.6 マージ + +| 項目 | 値 | +|------|-----| +| マージ実行者 | | +| マージ日 | | +| マージ先 | DD-005_session.md, ST-001_acceptance.md | + +--- + +## 4. 影響範囲 + +### 4.1 影響するドキュメント + +| ドキュメント | 変更種別 | 概要 | +|-------------|---------|------| +| docs/design/detail/DD-005_session.md | 修正 | §3.2 refresher 動作表・§7.1 実装状況を re-INVITE フォールバック対応に更新 | +| docs/test/system/ST-001_acceptance.md | 修正 | AC-3 Session Timer に UPDATE なし相手の re-INVITE ケースを追加 | + +### 4.2 影響するコード + +| モジュール | 変更種別 | 概要 | +|-----------|---------|------| +| `src/protocol/session/handlers/mod.rs` L662付近 | 修正 | `SessionRefreshDue` 処理から `update_session_expires` 呼び出しを削除。`SipSendSessionRefresh` を発行するだけにする | +| `src/protocol/session/handlers/timer_handler.rs` L32付近 | 修正 | `update_session_expires` の呼び出しタイミングを 2xx 受信時のみに変更（送信時点での呼び出し削除に伴う整合） | +| `src/main.rs` L601付近 | 修正 | `SessionOut::SipSendUpdate { expires }` を `SessionOut::SipSendSessionRefresh { expires, local_sdp }` に読み替え、`SipCommand::SendSessionRefresh { expires, local_sdp }` として SIP core へ転送する（`call_id` は既存どおり第1引数で渡す） | +| `src/protocol/sip/core.rs` L918付近 | 修正 | INVITE 受信時（`handle_invite`）に `Allow` ヘッダを解析し、`InviteContext` の `allow_update` フィールドに保持する | +| `src/protocol/sip/core.rs` L1695付近 | 修正/追加 | `SipSendSessionRefresh` 受信時に `allow_update` を参照し、UPDATE または `build_reinvite_request` を選択する | +| `src/protocol/sip/core.rs` L762付近 | 修正 | `handle_response` に outbound refresh の 2xx 受信処理（ACK 送信 + `SipEvent::SessionRefresh` emit）と 408/481 受信時の BYE + `SipEvent::SessionRefreshFailed` emit を追加 | +| `src/shared/ports/sip.rs` L38付近 | 追加 | `SipEvent::SessionRefreshFailed { call_id }` バリアントを追加 | +| `src/protocol/session/types.rs` L109付近 | 追加 | `SessionControlIn::SipSessionRefreshFailed { call_id }` バリアントを追加 | +| `src/main.rs` L487付近 | 修正 | `SipEvent::SessionRefreshFailed` を `SessionControlIn::SipSessionRefreshFailed` へルーティングする処理を追加 | + +--- + +## 5. 差分仕様（What / How） + +### 5.1 要件追加（DD-005_session.md §3.2 へマージ） + +```markdown +### 3.2 refresher の動作（更新後） + +| refresher | 責任 | Voicebot の動作 | +|-----------|------|----------------| +| `uas` | Voicebot | Session-Expires の 80% 経過時に session refresh を送信。相手の `Allow` に `UPDATE` が含まれる場合は `UPDATE`、それ以外（`Allow` 未受信・`UPDATE` 不在）は `re-INVITE` を送信する | +| `uac` | 相手側 | 相手からの re-INVITE/UPDATE を受信し 200 OK を返す（変更なし） | + +#### 判定ロジック（refresher=uas 時） + +1. **SIP core** は着信 INVITE 受信時（`handle_invite`）に相手の `Allow` ヘッダを解析し、`UPDATE` サポート有無を `InviteContext.allow_update: bool` として保持する（session 層は `allow_update` を持たない） +2. `Allow` ヘッダが存在しない、または `UPDATE` が含まれない場合は `allow_update = false` として扱う（安全側） +3. **session 層** は `SessionRefreshDue` 発火時に `SessionOut::SipSendSessionRefresh { expires, local_sdp: Option }` を発行するだけでよい（UPDATE か re-INVITE かは SIP core が決定する） +4. **SIP core** は `SipSendSessionRefresh` を受け取り、保持済みの `InviteContext.allow_update` に基づき送信方式を選択する: + - `allow_update == true` → UPDATE を送信（従来動作） + - `allow_update == false` → re-INVITE を送信（新規） +5. **タイマリセットは 2xx 受信時のみ行う**: `SessionRefreshDue` ハンドラ（`mod.rs` L669）の `update_session_expires` 呼び出しを削除する。SIP core が 2xx 受信後に既存の `SipEvent::SessionRefresh { call_id, timer }` を emit し、main.rs が既存の `SessionControlIn::SipSessionExpires { timer }` へ転送する（既存コードパスを再利用し、新規イベントは不要） +``` + +### 5.2 詳細設計追加（DD-005_session.md §7.1 更新 + core.rs） + +#### (A) SIP core の `InviteContext` への `allow_update` フィールド追加 + +```markdown +`InviteContext`（`src/protocol/sip/core.rs` L918付近）に以下を追加する: + +| フィールド | 型 | 初期値 | 説明 | +|-----------|-----|--------|------| +| `allow_update` | `bool` | `false` | ダイアログ相手が UPDATE をサポートするか。着信 INVITE の `Allow` ヘッダ解析結果を格納 | + +設定タイミング: +- UAS 動作: `handle_invite` で着信 INVITE の `Allow` ヘッダを解析し、`InviteContext` 生成時に設定する +- UAC 動作（将来）: 2xx の `Allow` ヘッダを解析する（本 issue スコープ外） + +session 層はこのフィールドを持たない。`allow_update` の参照と UPDATE/re-INVITE の選択は SIP core 内に閉じる。 +``` + +#### (B) `SessionOut` バリアント変更 + +```markdown +`SessionOut::SipSendUpdate { expires }` を `SessionOut::SipSendSessionRefresh { expires, local_sdp: Option }` に変更する。 + +変更の意図: +- session 層は UPDATE か re-INVITE かを知る必要がなく、「session refresh を送信してほしい」という要求だけを出す +- SIP core が `InviteContext.allow_update` に基づき UPDATE か `build_reinvite_request` かを選択する +- re-INVITE の body SDP は `InviteContext.final_ok_payload`（SIPメッセージ全体の再送バッファ）では取得できないため、session 層が保持する `self.local_sdp`（`coordinator.rs` L94、型は `Option`）をそのまま `local_sdp` フィールドに格納して渡す +- `SessionRefreshDue` ハンドラは `self.local_sdp.clone()` を `local_sdp` として設定し、`SipSendSessionRefresh` を発行する + +`local_sdp == None` の扱い: +- session 確立前（`local_sdp` が未設定）に `SessionRefreshDue` が発火するケースは通常ない（Confirmed 遷移後のみ timer が起動する）が、防御的に `None` の場合は SIP core がログ出力のみ行い送信をスキップする + +`main.rs` L601付近の既存パターンに合わせて以下のように変更する: + +``` +// 変更前 +SessionOut::SipSendUpdate { expires } => { + sip_core.handle_sip_command(&call_id, SipCommand::SendUpdate { expires }); +} +// 変更後 +SessionOut::SipSendSessionRefresh { expires, local_sdp } => { + sip_core.handle_sip_command(&call_id, SipCommand::SendSessionRefresh { expires, local_sdp }); +} +``` + +`call_id` は `handle_sip_command` の第1引数として渡すため、`SipCommand::SendSessionRefresh` には含めない（既存パターンに準拠）。 +``` + +#### (C) `SipCommand::SendSessionRefresh` 処理と re-INVITE 生成（sip/core.rs） + +```markdown +SIP core の `SendSessionRefresh` ハンドラで以下を行う: + +1. `call_id` から `InviteContext` を取得する +2. `InviteContext.allow_update` を参照する: + - `true` → 既存の `build_update_request` を呼ぶ（従来動作） + - `false` → `build_reinvite_request` を呼ぶ（新規） +3. **タイマリセットは行わない**（送信時点では `update_session_expires` を呼ばない） + +`build_reinvite_request(ctx: &InviteContext, local_sdp: &Sdp, expires: Duration) -> SipMessage` + +処理フロー: +1. `InviteContext` の From/To/Call-ID/Route を引き継ぎ INVITE リクエストを組み立てる +2. CSeq を `InviteContext.local_cseq + 1` にインクリメントする +3. `Session-Expires` ヘッダ（`refresher=uas`）を付与する +4. `local_sdp`（`SipSendSessionRefresh` で渡された session 層の `self.local_sdp`）を body に設定する + ※ `final_ok_payload` は 200 OK 再送用のSIPメッセージ全体であり SDP body 取得には使わない +5. 送信後、`InviteContext` に「refresh 応答待ち」フラグを立てる（UPDATE / re-INVITE 共通） + +エラーケース: +| エラー | 条件 | 対応 | +|--------|------|------| +| context 不存在 | `call_id` に対応する `InviteContext` が無い | ログ出力のみ | +| `local_sdp` 未設定 | `SipSendSessionRefresh` の `local_sdp` が `None` | ログ出力のみ・送信スキップ | +| CSeq オーバーフロー | CSeq が u32 上限 | ログ出力・BYE 送信 | +``` + +#### (D) outbound re-INVITE / UPDATE の応答処理（sip/core.rs handle_response） + +```markdown +`handle_session_refresh_response(status: u16, ctx: &mut InviteContext)` として実装する +（UPDATE・re-INVITE 共通。既存の `handle_response` 内の「re-INVITE 応答待ちフラグ」確認後に呼ぶ） + +処理フロー: +1. `InviteContext` の「refresh 応答待ちフラグ」を確認し、outbound refresh の応答として処理する +2. ステータスコードを確認する +3. `2xx` の場合（RFC 3261 §13.2.2.4 準拠）: + a. re-INVITE の場合のみ ACK を生成・送信する（UPDATE は ACK 不要） + b. `Session-Expires` ヘッダを解析しタイマ情報を取得する + c. `InviteContext.allow_update` は**更新しない**（初期 INVITE 受信時に設定した値を使い続ける。refresh 応答の `Allow` 再評価は本 issue スコープ外） + d. **既存の** `SipEvent::SessionRefresh { call_id, timer }` を emit する + → main.rs の既存ルーティングにより `SessionControlIn::SipSessionExpires { timer }` として session 層へ転送される + → session 層の `update_session_expires` はこの通知を受けて呼ぶ（送信時点では呼ばない） +4. `408` / `481` の場合（RFC 4028 §10 準拠。**transaction timeout は現行実装では観測不可のため対象外**）: + - BYE を送信し通話を終了する + - **新設** `SipEvent::SessionRefreshFailed { call_id }` を emit する + - main.rs に新設ルーティングを追加し、**新設** `SessionControlIn::SipSessionRefreshFailed` として session 層へ転送する + - session 層の `SipSessionRefreshFailed` ハンドラは BYE 送信後の後処理（リソース解放等）を行う +5. `422 Session Interval Too Small` の場合: + - Min-SE を応答から更新して再送する（既存ロジックに準じる） +6. その他の非 `2xx` 応答の場合（RFC 4028 §10 準拠）: + - 即時 BYE はしない + - session expiration は延長しない（タイマリセットを行わない = `SipEvent` を emit しない） + - 既存タイマの満了時に `SessionControlIn::SessionTimerFired` が発火し、session 層が `SessionOut::AppSessionTimeout` を emit する + - `AppSessionTimeout` を受け取った app 層（service/call_control）が `HangupRequested` を返し、session 層が BYE を送信する（DD-005_session.md §3.5 準拠） + - **実装変更不要**: 本 issue で追加する「非 2xx の場合は `SipEvent` を emit しない」ことで、既存の session timeout 終話フローがそのまま動作する + +> **NOTE**: RFC 4028 §10 の「transaction timeout」は client transaction timer（Timer B/F）の発火に相当するが、 +> 現行実装には outbound refresh 用の client transaction timeout 検知経路が存在しない。 +> client transaction 実装の追加は本 issue のスコープ外とし、 +> 「timeout」の実質的な救済は session 層の `SessionTimerFired` → `AppSessionTimeout` → app 層の `HangupRequested` → BYE（既存終話フロー）で代替する。 +``` + +### 5.3 テストケース追加（ST-001_acceptance.md AC-3 へマージ） + +```markdown +### AC-3: Session Timer（更新後） + +| # | シナリオ | 期待結果 | SIPp | +|---|---------|---------|------| +| AC-3.1 | Session-Expires 受信（Allow: UPDATE あり）・UPDATE 2xx 受信 | UPDATE で refresh → 2xx 受信後にタイマがリセットされる | basic_uas_update.xml（既存） | +| AC-3.2 | Min-SE 下回り | 422 + Min-SE: 90 | basic_uas_update.xml（既存） | +| AC-3.3 | Allow: UPDATE なし（refresher=uas） | Session-Expires 80% 経過後に re-INVITE が送信される | basic_uas_reinvite_refresh.xml（要作成） | +| AC-3.4 | Allow ヘッダなし（refresher=uas） | Session-Expires 80% 経過後に re-INVITE が送信される | basic_uas_reinvite_refresh.xml（共用可） | +| AC-3.5 | re-INVITE 2xx 受信 | ACK 送信 + タイマリセット（SIP core が `SipEvent::SessionRefresh` を emit し session 層が `SipSessionExpires` 経由で `update_session_expires` を呼ぶ） | basic_uas_reinvite_refresh.xml | +| AC-3.6 | re-INVITE 408 / 481 受信 | BYE を送信し通話終了する | basic_uas_reinvite_refresh.xml（エラーシナリオ） | +| AC-3.7 | re-INVITE その他非 2xx（例: 500） | 即時 BYE しない・session expiration は延長しない・`SessionTimerFired` 発火後 `AppSessionTimeout` 経由で app 層が終話する | basic_uas_reinvite_refresh.xml（エラーシナリオ） | + +> **スコープ外**: transaction timeout（Timer B/F 発火）の検知は client transaction 実装が必要なため本 issue のスコープ外。 +> timeout の実質的な救済は `SessionTimerFired` → `AppSessionTimeout` → app 層の `HangupRequested` → BYE（既存終話フロー）による AC-3.7 の挙動で代替する。 +> +> SIPp シナリオ `basic_uas_reinvite_refresh.xml` は Codex が Approved 後に作成する。 +> エラーシナリオは同ファイル内のバリアントとして追加してよい。 +``` + +--- + +## 6. トレーサビリティ + +| From | To | 関係 | +|------|-----|------| +| Issue #280 | STEER-280 | 起票 | +| STEER-280 | DD-005_session.md §3.2, §7.1 | 設計修正 | +| STEER-280 | ST-001_acceptance.md AC-3.3〜3.7 | 受入条件追加 | +| DD-005_session.md §3.2 | src/protocol/session/handlers/mod.rs | 実装 | +| DD-005_session.md §3.2 | src/protocol/sip/core.rs | 実装 | +| ST-001_acceptance.md AC-3.3〜3.7 | basic_uas_reinvite_refresh.xml | E2E テスト（Codex が作成） | + +--- + +## 7. レビューチェックリスト + +### 7.1 仕様レビュー（Review → Approved） + +- [ ] `Allow` なし相手を `allow_update = false`（安全側）として扱う方針に合意している（→ Q1 解決済み） +- [ ] `allow_update` が SIP core の `InviteContext` に置かれ、session 層には存在しないことを確認した +- [ ] outbound re-INVITE の SDP 元データが session 層の `local_sdp`（`coordinator.rs` L94）経由で `SipSendSessionRefresh.local_sdp` として渡されることを確認した（`final_ok_payload` は使わない） +- [ ] `local_sdp == None` の場合に SIP core が送信スキップ+ログ出力することを確認した +- [ ] re-INVITE の `2xx` に対する ACK 生成ロジックが RFC 3261 §13.2.2.4 を満たすか（UPDATE には ACK 不要であることを確認） +- [ ] タイマリセット（`update_session_expires`）が `SessionRefreshDue` ハンドラから削除され、2xx 受信後に SIP core が既存 `SipEvent::SessionRefresh` を emit することで session 層が `update_session_expires` を呼ぶことを確認した +- [ ] 408/481 の場合に新設 `SipEvent::SessionRefreshFailed` が emit され、main.rs で `SessionControlIn::SipSessionRefreshFailed` へルーティングされることを確認した +- [ ] その他非 2xx の場合はイベントを emit せず、既存の `SessionTimerFired` → `AppSessionTimeout` → `HangupRequested` → BYE のフローで終話することを確認した（→ Q2 解決済み） +- [ ] transaction timeout のスコープ外処理が AC-3.7 の `SessionTimerFired` で代替されることに合意しているか +- [ ] UAC 動作（Voicebot 発信）側への影響がないか確認した +- [ ] 既存 `UPDATE` パスを壊していないこと（AC-3.1 が引き続き通ること） +- [ ] トレーサビリティが維持されているか + +### 7.2 マージ前チェック（Approved → Merged） + +- [ ] 実装が完了している +- [ ] AC-3.1〜3.7 がすべて PASS +- [ ] `basic_uas_reinvite_refresh.xml` シナリオファイルが Codex により作成されている +- [ ] CodeRabbit レビューを受けている +- [ ] DD-005_session.md / ST-001_acceptance.md への本体反映準備ができている + +--- + +## 8. 備考 + +### 解決済みクエスチョン + +| # | 質問 | 決定者 | 回答 | +|---|------|--------|------| +| Q1 | `Allow` ヘッダなし相手を `allow_update = false` として扱う（安全側）でよいか？ | @MasanoriSuda | **解決**: Yes。RFC 3311 は UPDATE サポートの判断材料を `Allow: UPDATE` または mid-dialog OPTIONS とする。`Allow` なしは「未確認」のため安全側（`re-INVITE`）に倒す | +| Q2 | re-INVITE が非 2xx で失敗した場合の動作は？ | @MasanoriSuda | **解決**: RFC 4028 §10 準拠。408/481 は即 BYE。その他非 2xx は即 BYE せず session expiration を延長しない。期限到来時は `SessionTimerFired` → `AppSessionTimeout` → app 層の `HangupRequested` → BYE（既存フロー）。transaction timeout（Timer B/F）は client transaction 未実装のため本 issue スコープ外（`SessionTimerFired` 経由の既存終話フローで代替） | +| Q3 | SIPp テストシナリオ `basic_uas_reinvite_refresh.xml` の作成担当は？ | @MasanoriSuda | **解決**: Codex 担当。docs ではなく backend テスト資産のため Approved 後に Codex が作成する | + +### RFC 参照 + +- RFC 4028 §8: Behavior of a Refresher — "If the remote side does not support the UPDATE method... the refresher MUST use re-INVITE." +- RFC 4028 §10: Handling of Session Expiration — timeout/408/481 は BYE。その他の非 2xx は即 BYE しない。 +- RFC 3311: UPDATE method — `Allow` ヘッダまたは mid-dialog OPTIONS を UPDATE サポートの判断材料とする。 +- RFC 3261 §13.2.2.4: 2xx への ACK は UAC コア層（ダイアログ）の責務。 + +--- + +## 変更履歴 + +| 日付 | 変更内容 | 作成者 | +|------|---------|--------| +| 2026-03-10 | 初版作成 | Claude Sonnet 4.6 | +| 2026-03-10 | Q1〜Q3 解決反映: §5.2(D) を RFC 4028 §10 準拠エラー処理に更新、AC-3.6〜3.7 追加、§8 クエスチョン解決 | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映: ①`allow_update` を SIP core の `InviteContext` に移設（session 境界整理）、②タイマリセットを 2xx 受信時のみに変更（`SessionRefreshDue` からの `update_session_expires` 削除）、③ transaction timeout をスコープ外とし AC-3.6 を 408/481 に限定 | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映（第2回）: ①SDP 元データを `final_ok_payload` → session 層の `local_sdp` に修正し `SipSendSessionRefresh` に追加、②`SessionIn` を実際の型（`SipEvent`/`SessionControlIn`）に置換・2xx 成功は既存 `SipEvent::SessionRefresh` 再利用・失敗は新設 `SipEvent::SessionRefreshFailed`、③AC-3.1 に UPDATE 2xx 後タイマリセット検証を追加 | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映（第3回）: ①受入条件 L50 を `408/481`（transaction timeout 除外）に揃え、②`local_sdp` 型を `Option` に統一し `None` 時の挙動を明記、③§7.1 チェックリストを `local_sdp`・`SipEvent::SessionRefresh` 再利用・`SipEvent::SessionRefreshFailed` ルーティングに更新 | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映（第4回）: ①§5.1 判定ロジックの `SipSendSessionRefresh { call_id }` を `{ expires, local_sdp }` に揃え、②`SipCommand::SendSessionRefresh` から `call_id` を除外（既存パターン準拠・第1引数渡し）、③AC-3.5 の `SessionRefreshed` を `SipEvent::SessionRefresh` / `SipSessionExpires` 経由に更新 | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映（第5回）: §5.2(D) の refresh 2xx 処理で `allow_update` を再評価しないことを明記（初期 INVITE 受信時の値を維持。§5.2(A) スコープ外宣言と整合） | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映（第6回）: `SessionTimerFired` が直接 BYE しないことを実装確認し、「AppSessionTimeout → HangupRequested → BYE」の実際のフローに修正（受入条件 L51・AC-3.7・Q2・§5.2(D) step 6・§7.1 チェックリスト）。実装変更不要であることも明記 | Claude Sonnet 4.6 | +| 2026-03-10 | Codex 指摘反映（第7回）: §5.2(D) NOTE および AC-3.7 スコープ外 NOTE の `SessionTimerFired`「期限切れ BYE」文言を `SessionTimerFired → AppSessionTimeout → HangupRequested → BYE（既存終話フロー）` に修正。Codex レビュー OK 確認 | Claude Sonnet 4.6 | diff --git a/virtual-voicebot-backend/src/main.rs b/virtual-voicebot-backend/src/main.rs index 0d5d7bc..6d46df5 100644 --- a/virtual-voicebot-backend/src/main.rs +++ b/virtual-voicebot-backend/src/main.rs @@ -492,6 +492,16 @@ async fn main() -> anyhow::Result<()> { .await; } } + SipEvent::SessionRefreshFailed { call_id } => { + if let Some(sess_tx) = session_registry.get(&call_id).await { + let _ = sess_tx + .control_tx + .send(SessionControlIn::SipSessionRefreshFailed { + call_id: call_id.clone(), + }) + .await; + } + } SipEvent::TransactionTimeout { call_id } => { log::warn!("[main] TransactionTimeout for call_id={}", call_id); if let Some(sess_tx) = session_registry.get(&call_id).await { @@ -598,8 +608,11 @@ async fn main() -> anyhow::Result<()> { SessionOut::SipSend200 { answer } => { sip_core.handle_sip_command(&call_id, SipCommand::Send200 { answer }); } - SessionOut::SipSendUpdate { expires } => { - sip_core.handle_sip_command(&call_id, SipCommand::SendUpdate { expires }); + SessionOut::SipSendSessionRefresh { expires, local_sdp } => { + sip_core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { expires, local_sdp }, + ); } SessionOut::SipSendError { code, reason } => { sip_core.handle_sip_command( diff --git a/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs b/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs index e1a09f4..1e514c7 100644 --- a/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs +++ b/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs @@ -13,7 +13,6 @@ use crate::protocol::rtp::codec::mulaw_to_linear16; use crate::protocol::session::b2bua; use crate::protocol::session::types::{ IvrState, SessState, SessionControlIn, SessionMediaIn, SessionOut, SessionRefresher, - SessionTimerInfo, }; use crate::service::routing::{ActionConfig, ActionExecutor, RuleEvaluator}; use crate::shared::ports::app::{AppEvent, EndReason, RtpAudioChunk}; @@ -639,6 +638,28 @@ impl SessionCoordinator { (_, SessionControlIn::SipSessionExpires { timer }) => { self.update_session_expires(timer); } + (_, SessionControlIn::SipSessionRefreshFailed { call_id: _ }) => { + warn!( + "[session {}] session refresh failed after BYE, cleaning up session", + self.call_id + ); + self.stop_ring_delay(); + self.cancel_transfer(); + self.shutdown_b_leg(true).await; + self.cancel_playback(); + self.stop_keepalive_timer(); + self.stop_session_timer(); + self.stop_ivr_timeout(); + self.mark_transfer_ended(); + self.stop_recorders(); + self.send_ingest("ended").await; + self.rtp.stop(self.call_id.as_str()); + let _ = self + .session_out_tx + .send((self.call_id.clone(), SessionOut::RtpStopTx)) + .await; + self.send_call_ended(EndReason::Timeout); + } (_, SessionControlIn::IvrTimeout) => { if self.ivr_state == IvrState::IvrMenuWaiting { if self.ivr_keypad_node_id.is_some() { @@ -663,13 +684,13 @@ impl SessionCoordinator { if let (Some(expires), Some(SessionRefresher::Uas)) = (self.session_expires, self.session_refresher) { - let _ = self - .session_out_tx - .try_send((self.call_id.clone(), SessionOut::SipSendUpdate { expires })); - self.update_session_expires(SessionTimerInfo { - expires, - refresher: SessionRefresher::Uas, - }); + let _ = self.session_out_tx.try_send(( + self.call_id.clone(), + SessionOut::SipSendSessionRefresh { + expires, + local_sdp: self.local_sdp.clone(), + }, + )); } } (_, SessionControlIn::SessionTimerFired) => { @@ -1484,7 +1505,7 @@ mod tests { use crate::shared::config::{ OutboundConfig, RegistrarConfig, RegistrarTransport, SessionRuntimeConfig, }; - use crate::shared::ports::app::app_event_channel; + use crate::shared::ports::app::{app_event_channel, AppEvent, AppEventRx, EndReason}; use crate::shared::ports::call_log_port::{CallLogPort, EndedCallLog}; use crate::shared::ports::ingest::{IngestError, IngestFuture, IngestPayload, IngestPort}; use crate::shared::ports::routing_port::{ @@ -1497,7 +1518,7 @@ mod tests { use std::net::SocketAddr; use std::sync::{Arc, Mutex}; use tokio::sync::mpsc; - use tokio::time::Duration; + use tokio::time::{timeout, Duration}; struct DummyIngestPort; @@ -1690,12 +1711,17 @@ mod tests { } } - fn build_test_session( + fn build_test_session_with_observers( routing_port: Arc, - ) -> (SessionCoordinator, mpsc::Receiver<(CallId, SessionOut)>) { + ) -> ( + SessionCoordinator, + mpsc::Receiver<(CallId, SessionOut)>, + AppEventRx, + mpsc::Receiver, + ) { let (session_out_tx, session_out_rx) = mpsc::channel(32); - let (app_tx, _app_rx) = app_event_channel(16); - let (control_tx, _control_rx) = + let (app_tx, app_rx) = app_event_channel(16); + let (control_tx, control_rx) = mpsc::channel(super::super::SESSION_CONTROL_CHANNEL_CAPACITY); let (media_tx, _media_rx) = mpsc::channel(super::super::SESSION_MEDIA_CHANNEL_CAPACITY); let base_cfg = crate::shared::config::Config::from_env().expect("config loads"); @@ -1791,6 +1817,14 @@ mod tests { dtmf_history: Vec::new(), notification_sent: false, }; + (session, session_out_rx, app_rx, control_rx) + } + + fn build_test_session( + routing_port: Arc, + ) -> (SessionCoordinator, mpsc::Receiver<(CallId, SessionOut)>) { + let (session, session_out_rx, _app_rx, _control_rx) = + build_test_session_with_observers(routing_port); (session, session_out_rx) } @@ -2441,4 +2475,82 @@ mod tests { ); assert!(session.playback.is_some()); } + + #[tokio::test] + async fn session_refresh_due_emits_session_refresh_request_with_local_sdp() { + let routing_port = Arc::new(NoopRoutingPort::new()); + let (mut session, mut session_out_rx) = build_test_session(routing_port); + session.local_sdp = Some(Sdp::pcmu("127.0.0.1", 10000)); + session.session_expires = Some(Duration::from_secs(90)); + session.session_refresher = Some(SessionRefresher::Uas); + + let advance = session + .handle_control_event(SessState::Established, SessionControlIn::SessionRefreshDue) + .await; + + assert!(advance); + let (_call_id, out) = session_out_rx.try_recv().expect("session refresh output"); + match out { + SessionOut::SipSendSessionRefresh { expires, local_sdp } => { + assert_eq!(expires, Duration::from_secs(90)); + assert!(matches!( + local_sdp, + Some(sdp) if sdp.ip == "127.0.0.1" && sdp.port == 10000 + )); + } + other => panic!("expected SipSendSessionRefresh, got {:?}", other), + } + } + + #[tokio::test] + async fn sip_session_refresh_failed_stops_timers_and_emits_timeout_end() { + let routing_port = Arc::new(NoopRoutingPort::new()); + let (mut session, mut session_out_rx, app_rx, mut control_rx) = + build_test_session_with_observers(routing_port); + session.started_at = Some(tokio::time::Instant::now()); + session + .timers + .start_keepalive(session.control_tx.clone(), Duration::from_millis(100)); + session.timers.start_session_timer( + session.control_tx.clone(), + Duration::from_millis(120), + Some(Duration::from_millis(80)), + ); + + let call_id = session.call_id.clone(); + let advance = session + .handle_control_event( + SessState::Established, + SessionControlIn::SipSessionRefreshFailed { + call_id: call_id.clone(), + }, + ) + .await; + + assert!(advance); + + let (emitted_call_id, out) = session_out_rx.recv().await.expect("session output"); + assert_eq!(emitted_call_id, call_id); + assert!(matches!(out, SessionOut::RtpStopTx)); + + let app_event = timeout(Duration::from_millis(50), app_rx.recv()) + .await + .expect("call ended should be emitted") + .expect("app event"); + assert!(matches!( + app_event, + AppEvent::CallEnded { + call_id: ended_call_id, + reason: EndReason::Timeout, + .. + } if ended_call_id == call_id + )); + + assert!( + timeout(Duration::from_millis(180), control_rx.recv()) + .await + .is_err(), + "keepalive/session/refresh timers should be stopped after refresh failure" + ); + } } diff --git a/virtual-voicebot-backend/src/protocol/session/types.rs b/virtual-voicebot-backend/src/protocol/session/types.rs index 0a9963c..5a9c5af 100644 --- a/virtual-voicebot-backend/src/protocol/session/types.rs +++ b/virtual-voicebot-backend/src/protocol/session/types.rs @@ -110,6 +110,10 @@ pub enum SessionControlIn { SipSessionExpires { timer: SessionTimerInfo, }, + /// outbound session refresh が失敗し、SIP core 側で BYE を送信済み + SipSessionRefreshFailed { + call_id: CallId, + }, Abort(SessionError), } @@ -194,9 +198,10 @@ pub enum SessionOut { SipSend200 { answer: Sdp, }, - /// SIP UPDATE によるセッションリフレッシュ - SipSendUpdate { + /// SIP UPDATE / re-INVITE によるセッションリフレッシュ + SipSendSessionRefresh { expires: Duration, + local_sdp: Option, }, /// SIP エラー応答（INVITE の最終応答） SipSendError { diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index 94cd58e..73ce20e 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -46,6 +46,8 @@ struct InviteContext { session_timer: Option, final_ok: Option, final_ok_payload: Option>, + allow_update: bool, + pending_refresh: Option, local_cseq: u32, expires_at: Option, } @@ -65,6 +67,36 @@ struct SessionTimerConfig { min_se: u64, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SessionRefreshMethod { + Update, + Invite, +} + +impl SessionRefreshMethod { + fn as_sip_method(self) -> SipMethod { + match self { + Self::Update => SipMethod::Update, + Self::Invite => SipMethod::Invite, + } + } + + fn as_cseq_method(self) -> &'static str { + match self { + Self::Update => "UPDATE", + Self::Invite => "INVITE", + } + } +} + +#[derive(Debug, Clone)] +struct PendingSessionRefresh { + method: SessionRefreshMethod, + cseq: u32, + expires: Duration, + local_sdp: Option, +} + #[derive(Debug, Clone)] struct RAckHeader { rseq: u32, @@ -193,6 +225,13 @@ fn header_has_token(value: Option<&str>, token: &str) -> bool { .any(|part| part.eq_ignore_ascii_case(token)) } +fn response_header_value<'a>(resp: &'a SipResponse, name: &str) -> Option<&'a str> { + resp.headers + .iter() + .find(|header| header.name.eq_ignore_ascii_case(name)) + .map(|header| header.value.as_str()) +} + fn supports_100rel(req: &SipRequest) -> bool { header_has_token(req.header_value("Supported"), "100rel") || header_has_token(req.header_value("Require"), "100rel") @@ -351,19 +390,93 @@ fn build_update_request( ctx: &mut InviteContext, cfg: &SipConfig, expires: Duration, -) -> Option> { +) -> Result, SessionRefreshBuildError> { + build_session_refresh_request(ctx, cfg, SessionRefreshMethod::Update, expires, None) +} + +fn build_reinvite_request( + ctx: &mut InviteContext, + cfg: &SipConfig, + local_sdp: &Sdp, + expires: Duration, +) -> Result, SessionRefreshBuildError> { + build_session_refresh_request( + ctx, + cfg, + SessionRefreshMethod::Invite, + expires, + Some(local_sdp), + ) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SessionRefreshBuildError { + MissingHeaders, + MissingLocalSdp, + CseqOverflow, +} + +fn next_local_cseq(ctx: &mut InviteContext) -> Result { + let next = if ctx.local_cseq == 0 { + 1 + } else { + ctx.local_cseq + .checked_add(1) + .ok_or(SessionRefreshBuildError::CseqOverflow)? + }; + ctx.local_cseq = next; + Ok(next) +} + +fn sdp_body(sdp: &Sdp) -> Vec { + format!( + concat!( + "v=0\r\n", + "o=rustbot 1 1 IN IP4 {ip}\r\n", + "s=Rust PCMU Bot\r\n", + "c=IN IP4 {ip}\r\n", + "t=0 0\r\n", + "m=audio {rtp} RTP/AVP {pt}\r\n", + "a=rtpmap:{pt} {codec}\r\n", + "a=sendrecv\r\n", + ), + ip = sdp.ip, + rtp = sdp.port, + pt = sdp.payload_type, + codec = sdp.codec + ) + .into_bytes() +} + +fn build_session_refresh_request( + ctx: &mut InviteContext, + cfg: &SipConfig, + method: SessionRefreshMethod, + expires: Duration, + local_sdp: Option<&Sdp>, +) -> Result, SessionRefreshBuildError> { let req = &ctx.req; - let to = req.header_value("From")?.to_string(); - let mut from = req.header_value("To")?.to_string(); + let to = req + .header_value("From") + .ok_or(SessionRefreshBuildError::MissingHeaders)? + .to_string(); + let mut from = req + .header_value("To") + .ok_or(SessionRefreshBuildError::MissingHeaders)? + .to_string(); if !from.to_ascii_lowercase().contains("tag=") { from = format!("{from};tag=rustbot"); } - let call_id = req.header_value("Call-ID")?.to_string(); + let call_id = req + .header_value("Call-ID") + .ok_or(SessionRefreshBuildError::MissingHeaders)? + .to_string(); let uri = req .header_value("Contact") .map(extract_contact_uri) .unwrap_or_else(|| req.uri.as_str()) .to_string(); + let req_uri = req.uri.clone(); let transport = match ctx.tx.peer { TransportPeer::Udp(_) => "UDP", TransportPeer::Tcp(_) => "TCP", @@ -375,17 +488,16 @@ fn build_update_request( cfg.sip_port, generate_branch() ); - let cseq = ctx.local_cseq.saturating_add(1).max(1); - ctx.local_cseq = cseq; + let cseq = next_local_cseq(ctx)?; - let contact_scheme = contact_scheme_from_uri(&req.uri); - let builder = SipRequestBuilder::new(SipMethod::Update, uri) + let contact_scheme = contact_scheme_from_uri(&req_uri); + let mut builder = SipRequestBuilder::new(method.as_sip_method(), uri) .header("Via", via) .header("Max-Forwards", "70") .header("From", from) .header("To", to) .header("Call-ID", call_id) - .header("CSeq", format!("{cseq} UPDATE")) + .header("CSeq", format!("{cseq} {}", method.as_cseq_method())) .header( "Contact", format!( @@ -393,9 +505,57 @@ fn build_update_request( cfg.advertised_ip, cfg.sip_port ), ) - .header("Session-Expires", expires.as_secs().to_string()) + .header( + "Session-Expires", + format!( + "{};refresher={}", + expires.as_secs(), + SessionRefresher::Uas.as_str() + ), + ) .header("Supported", "timer"); - Some(builder.build().to_bytes()) + if method == SessionRefreshMethod::Invite { + let local_sdp = local_sdp.ok_or(SessionRefreshBuildError::MissingLocalSdp)?; + builder = builder.body(sdp_body(local_sdp), Some("application/sdp")); + } + Ok(builder.build().to_bytes()) +} + +fn build_ack_request(ctx: &InviteContext, cfg: &SipConfig, invite_cseq: u32) -> Option> { + let req = &ctx.req; + let to = req.header_value("From")?.to_string(); + let mut from = req.header_value("To")?.to_string(); + if !from.to_ascii_lowercase().contains("tag=") { + from = format!("{from};tag=rustbot"); + } + let call_id = req.header_value("Call-ID")?.to_string(); + let uri = req + .header_value("Contact") + .map(extract_contact_uri) + .unwrap_or_else(|| req.uri.as_str()) + .to_string(); + let transport = match ctx.tx.peer { + TransportPeer::Udp(_) => "UDP", + TransportPeer::Tcp(_) => "TCP", + }; + let via = format!( + "SIP/2.0/{} {}:{};branch={}", + transport, + cfg.advertised_ip, + cfg.sip_port, + generate_branch() + ); + Some( + SipRequestBuilder::new(SipMethod::Ack, uri) + .header("Via", via) + .header("Max-Forwards", "70") + .header("From", from) + .header("To", to) + .header("Call-ID", call_id) + .header("CSeq", format!("{invite_cseq} ACK")) + .build() + .to_bytes(), + ) } fn build_bye_request(ctx: &mut InviteContext, cfg: &SipConfig) -> Option> { @@ -435,6 +595,36 @@ fn build_bye_request(ctx: &mut InviteContext, cfg: &SipConfig) -> Option Some(builder.build().to_bytes()) } +fn session_timer_info_from_response( + resp: &SipResponse, + fallback: Option<&SessionTimerConfig>, +) -> Option { + if let Some(value) = response_header_value(resp, "Session-Expires") { + let (expires, refresher_opt) = parse_session_expires(value)?; + return Some(SessionTimerInfo { + expires: Duration::from_secs(expires), + refresher: refresher_opt + .or_else(|| fallback.map(|cfg| cfg.refresher)) + .unwrap_or(SessionRefresher::Uas), + }); + } + + fallback.map(|cfg| SessionTimerInfo { + expires: cfg.expires, + refresher: cfg.refresher, + }) +} + +#[derive(Debug)] +struct SessionRefreshResponseAction { + call_id: CallId, + ack_payload: Option<(TransportPeer, Vec)>, + retry_payload: Option<(TransportPeer, Vec)>, + bye_payload: Option<(TransportPeer, Vec)>, + events: Vec, + remove_invite: bool, +} + /// Extracts the URI portion from a Contact header value. /// /// If the header contains an angle-bracketed address, returns the substring inside the first `<...>`. @@ -786,18 +976,30 @@ impl SipCore { return vec![]; } } - let call_id = resp - .headers - .iter() - .find(|header| header.name.eq_ignore_ascii_case("Call-ID")) - .map(|header| header.value.as_str()) - .unwrap_or("-"); - let cseq = resp - .headers - .iter() - .find(|header| header.name.eq_ignore_ascii_case("CSeq")) - .map(|header| header.value.as_str()) - .unwrap_or("-"); + if let Some(call_id) = response_header_value(&resp, "Call-ID") + .and_then(|value| CallId::new(value.to_string()).ok()) + { + if let Some(action) = self.handle_session_refresh_response(&call_id, &resp) { + if let Some((peer, payload)) = action.ack_payload { + self.send_payload(peer, payload); + } + if let Some((peer, payload)) = action.retry_payload { + self.send_payload(peer, payload); + } + if let Some((peer, payload)) = action.bye_payload { + self.send_payload(peer, payload); + } + if action.remove_invite { + if self.outbound_call_id.as_ref() == Some(&action.call_id) { + self.outbound_call_id = None; + } + self.invites.remove(&action.call_id); + } + return action.events; + } + } + let call_id = response_header_value(&resp, "Call-ID").unwrap_or("-"); + let cseq = response_header_value(&resp, "CSeq").unwrap_or("-"); log::warn!( "[sip] unhandled response status={} call_id={} cseq={} peer={:?}", resp.status_code, @@ -808,6 +1010,140 @@ impl SipCore { vec![SipEvent::Unknown] } + fn handle_session_refresh_response( + &mut self, + call_id: &CallId, + resp: &SipResponse, + ) -> Option { + let cseq = + response_header_value(resp, "CSeq").and_then(|value| parse_cseq_header(value).ok())?; + let ctx = self.invites.get_mut(call_id)?; + let pending = ctx.pending_refresh.clone()?; + if cseq.num != pending.cseq + || !cseq + .method + .eq_ignore_ascii_case(pending.method.as_cseq_method()) + { + return None; + } + + let peer = ctx.tx.peer; + let mut action = SessionRefreshResponseAction { + call_id: call_id.clone(), + ack_payload: None, + retry_payload: None, + bye_payload: None, + events: vec![], + remove_invite: false, + }; + if resp.status_code < 200 { + return Some(action); + } + if pending.method == SessionRefreshMethod::Invite && resp.status_code >= 200 { + action.ack_payload = + build_ack_request(ctx, &self.cfg, pending.cseq).map(|payload| (peer, payload)); + } + + match resp.status_code { + 200..=299 => { + ctx.pending_refresh = None; + if let Some(timer) = + session_timer_info_from_response(resp, ctx.session_timer.as_ref()) + { + if let Some(session_timer) = ctx.session_timer.as_mut() { + session_timer.expires = timer.expires; + session_timer.refresher = timer.refresher; + } + action.events.push(SipEvent::SessionRefresh { + call_id: call_id.clone(), + timer, + }); + } else { + log::warn!( + "[sip refresh] 2xx without usable Session-Expires call_id={}", + call_id + ); + } + } + 422 => { + let Some(min_se) = response_header_value(resp, "Min-SE").and_then(parse_min_se) + else { + ctx.pending_refresh = None; + log::warn!("[sip refresh] 422 without valid Min-SE call_id={}", call_id); + return Some(action); + }; + if let Some(session_timer) = ctx.session_timer.as_mut() { + session_timer.min_se = session_timer.min_se.max(min_se); + } + let retry_expires = Duration::from_secs(pending.expires.as_secs().max(min_se)); + let retry_result = match pending.method { + SessionRefreshMethod::Update => { + build_update_request(ctx, &self.cfg, retry_expires) + } + SessionRefreshMethod::Invite => { + let Some(local_sdp) = pending.local_sdp.as_ref() else { + ctx.pending_refresh = None; + log::warn!( + "[sip refresh] missing local SDP for 422 re-INVITE retry call_id={}", + call_id + ); + return Some(action); + }; + build_reinvite_request(ctx, &self.cfg, local_sdp, retry_expires) + } + }; + match retry_result { + Ok(payload) => { + ctx.pending_refresh = Some(PendingSessionRefresh { + method: pending.method, + cseq: ctx.local_cseq, + expires: retry_expires, + local_sdp: pending.local_sdp.clone(), + }); + action.retry_payload = Some((peer, payload)); + } + Err(SessionRefreshBuildError::MissingLocalSdp) => { + ctx.pending_refresh = None; + log::warn!( + "[sip refresh] missing local SDP for 422 retry call_id={}", + call_id + ); + } + Err(SessionRefreshBuildError::MissingHeaders) => { + ctx.pending_refresh = None; + log::warn!( + "[sip refresh] missing dialog headers for 422 retry call_id={}", + call_id + ); + } + Err(SessionRefreshBuildError::CseqOverflow) => { + ctx.pending_refresh = None; + action.bye_payload = + build_bye_request(ctx, &self.cfg).map(|payload| (peer, payload)); + action.events.push(SipEvent::SessionRefreshFailed { + call_id: call_id.clone(), + }); + action.remove_invite = true; + } + } + } + 408 | 481 => { + ctx.pending_refresh = None; + action.bye_payload = + build_bye_request(ctx, &self.cfg).map(|payload| (peer, payload)); + action.events.push(SipEvent::SessionRefreshFailed { + call_id: call_id.clone(), + }); + action.remove_invite = true; + } + _ => { + ctx.pending_refresh = None; + } + } + + Some(action) + } + /// Process an incoming INVITE request, handling retransmits, re-INVITEs, busy call rejection, /// and session-timer validation, and start a new invite server transaction when appropriate. /// @@ -924,6 +1260,8 @@ impl SipCore { session_timer: session_timer.clone(), final_ok: None, final_ok_payload: None, + allow_update: header_has_token(req.header_value("Allow"), "UPDATE"), + pending_refresh: None, local_cseq: 0, expires_at: None, }; @@ -1023,6 +1361,7 @@ impl SipCore { ctx.last_rseq = None; ctx.final_ok = None; ctx.final_ok_payload = None; + ctx.pending_refresh = None; if let Some(cfg) = &session_timer { ctx.session_timer = Some(cfg.clone()); } @@ -1692,18 +2031,83 @@ impl SipCore { ); } } - SipCommand::SendUpdate { expires } => { - let (peer, payload) = if let Some(ctx) = self.invites.get_mut(call_id) { + SipCommand::SendSessionRefresh { expires, local_sdp } => { + let result = if let Some(ctx) = self.invites.get_mut(call_id) { let peer = ctx.tx.peer; - let payload = build_update_request(ctx, &self.cfg, expires); - (Some(peer), payload) + let method = if ctx.allow_update { + SessionRefreshMethod::Update + } else { + SessionRefreshMethod::Invite + }; + let payload = match method { + SessionRefreshMethod::Update => { + build_update_request(ctx, &self.cfg, expires) + } + SessionRefreshMethod::Invite => { + let Some(local_sdp) = local_sdp.as_ref() else { + log::warn!( + "[sip refresh] local SDP missing, cannot send re-INVITE call_id={}", + call_id + ); + return; + }; + build_reinvite_request(ctx, &self.cfg, local_sdp, expires) + } + }; + Some((peer, method, payload)) } else { - (None, None) + None }; - if let (Some(peer), Some(payload)) = (peer, payload) { - self.send_payload(peer, payload); + if let Some((peer, method, payload)) = result { + match payload { + Ok(payload) => { + if let Some(ctx) = self.invites.get_mut(call_id) { + ctx.pending_refresh = Some(PendingSessionRefresh { + method, + cseq: ctx.local_cseq, + expires, + local_sdp, + }); + } + self.send_payload(peer, payload); + } + Err(SessionRefreshBuildError::MissingHeaders) => { + log::warn!( + "[sip refresh] failed to build {:?} request due to missing headers call_id={}", + method, + call_id + ); + } + Err(SessionRefreshBuildError::MissingLocalSdp) => { + log::warn!( + "[sip refresh] failed to build {:?} request due to missing local SDP call_id={}", + method, + call_id + ); + } + Err(SessionRefreshBuildError::CseqOverflow) => { + log::warn!( + "[sip refresh] CSeq overflow while building {:?} call_id={}", + method, + call_id + ); + let bye = if let Some(ctx) = self.invites.get_mut(call_id) { + build_bye_request(ctx, &self.cfg) + .map(|payload| (ctx.tx.peer, payload)) + } else { + None + }; + if let Some((peer, payload)) = bye { + self.send_payload(peer, payload); + } + if self.outbound_call_id.as_ref() == Some(call_id) { + self.outbound_call_id = None; + } + self.invites.remove(call_id); + } + } } else { - log::warn!("[sip update] failed to build UPDATE call_id={}", call_id); + log::warn!("[sip refresh] invite context missing call_id={}", call_id); } } SipCommand::SendError { code, reason } => { @@ -1874,6 +2278,58 @@ mod tests { TransportPeer::Udp("127.0.0.1:5060".parse().unwrap()) } + fn test_core() -> (SipCore, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(16); + let core = SipCore::new( + SipConfig { + advertised_ip: "127.0.0.1".to_string(), + sip_port: 5060, + advertised_rtp_port: 4000, + }, + tx, + ); + (core, rx) + } + + fn inbound_invite_request(call_id: &str, allow: Option<&str>) -> SipRequest { + let mut builder = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", "") + .header("Call-ID", call_id) + .header("CSeq", "1 INVITE"); + if let Some(allow) = allow { + builder = builder.header("Allow", allow); + } + builder.build() + } + + fn send_inbound_invite(core: &mut SipCore, call_id: &str, allow: Option<&str>) -> CallId { + let req = inbound_invite_request(call_id, allow); + let input = SipInput { + peer: dummy_peer(), + data: req.to_bytes(), + }; + let events = core.handle_input(&input); + assert!(matches!( + events.as_slice(), + [SipEvent::IncomingInvite { .. }] + )); + CallId::new(call_id.to_string()).expect("valid call id") + } + + fn parse_sent_request( + rx: &mut mpsc::Receiver, + ) -> (TransportPeer, SipRequest) { + let sent = rx.try_recv().expect("sent request"); + let text = String::from_utf8(sent.payload).expect("utf8 request"); + let req = match parse_sip_message(&text).expect("parse request") { + SipMessage::Request(req) => req, + other => panic!("expected request, got {:?}", other), + }; + (sent.peer, req) + } + fn dummy_invite_context() -> InviteContext { let req = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com").build(); InviteContext { @@ -1885,11 +2341,230 @@ mod tests { session_timer: None, final_ok: None, final_ok_payload: None, + allow_update: false, + pending_refresh: None, local_cseq: 0, expires_at: None, } } + #[test] + fn invite_records_allow_update_support_safely() { + for (call_id, allow, expected) in [ + ("call-allow-update", Some("INVITE, ACK, UPDATE"), true), + ("call-no-update", Some("INVITE, ACK, BYE"), false), + ("call-no-allow", None, false), + ] { + let (mut core, _rx) = test_core(); + let call_id = send_inbound_invite(&mut core, call_id, allow); + assert_eq!( + core.invites + .get(&call_id) + .expect("invite context") + .allow_update, + expected + ); + } + } + + #[test] + fn session_refresh_uses_reinvite_when_update_not_allowed() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite(&mut core, "call-reinvite", None); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + + let (_peer, req) = parse_sent_request(&mut rx); + match req.method { + SipMethod::Invite => {} + other => panic!("expected INVITE, got {:?}", other), + } + assert_eq!( + req.header_value("Session-Expires"), + Some("90;refresher=uas") + ); + assert_eq!(req.header_value("Content-Type"), Some("application/sdp")); + assert!(!req.body.is_empty(), "re-INVITE should carry SDP"); + + let pending = core + .invites + .get(&call_id) + .and_then(|ctx| ctx.pending_refresh.as_ref()) + .expect("pending refresh"); + assert_eq!(pending.method, SessionRefreshMethod::Invite); + } + + #[test] + fn session_refresh_uses_update_when_peer_supports_it() { + let (mut core, mut rx) = test_core(); + let call_id = + send_inbound_invite(&mut core, "call-update", Some("INVITE, ACK, UPDATE, BYE")); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: None, + }, + ); + + let (_peer, req) = parse_sent_request(&mut rx); + match req.method { + SipMethod::Update => {} + other => panic!("expected UPDATE, got {:?}", other), + } + assert_eq!( + req.header_value("Session-Expires"), + Some("120;refresher=uas") + ); + assert!(req.body.is_empty(), "UPDATE refresh should not carry SDP"); + + let pending = core + .invites + .get(&call_id) + .and_then(|ctx| ctx.pending_refresh.as_ref()) + .expect("pending refresh"); + assert_eq!(pending.method, SessionRefreshMethod::Update); + } + + #[test] + fn reinvite_refresh_2xx_sends_ack_and_emits_session_refresh() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite(&mut core, "call-refresh-ok", None); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + let (_peer, outbound_req) = parse_sent_request(&mut rx); + let outbound_cseq = outbound_req + .header_value("CSeq") + .expect("outbound CSeq") + .to_string(); + + let resp = SipResponseBuilder::new(200, "OK") + .header("Call-ID", "call-refresh-ok") + .header("CSeq", outbound_cseq) + .header("Session-Expires", "180;refresher=uas") + .build(); + let events = core.handle_response(resp, dummy_peer()); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-refresh-ok" + && timer.expires == Duration::from_secs(180) + && timer.refresher == SessionRefresher::Uas + )); + assert!(core + .invites + .get(&call_id) + .expect("invite context") + .pending_refresh + .is_none()); + + let (_peer, ack) = parse_sent_request(&mut rx); + match ack.method { + SipMethod::Ack => {} + other => panic!("expected ACK, got {:?}", other), + } + assert_eq!(ack.header_value("CSeq"), Some("1 ACK")); + } + + #[test] + fn reinvite_refresh_408_sends_ack_then_bye_and_emits_failure() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite(&mut core, "call-refresh-408", None); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + let (_peer, outbound_req) = parse_sent_request(&mut rx); + let outbound_cseq = outbound_req + .header_value("CSeq") + .expect("outbound CSeq") + .to_string(); + + let resp = SipResponseBuilder::new(408, "Request Timeout") + .header("Call-ID", "call-refresh-408") + .header("CSeq", outbound_cseq) + .build(); + let events = core.handle_response(resp, dummy_peer()); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefreshFailed { call_id }] if call_id.as_str() == "call-refresh-408" + )); + assert!(!core.invites.contains_key(&call_id)); + + let (_peer, ack) = parse_sent_request(&mut rx); + match ack.method { + SipMethod::Ack => {} + other => panic!("expected ACK, got {:?}", other), + } + let (_peer, bye) = parse_sent_request(&mut rx); + match bye.method { + SipMethod::Bye => {} + other => panic!("expected BYE, got {:?}", other), + } + } + + #[test] + fn reinvite_refresh_other_failure_does_not_send_bye() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite(&mut core, "call-refresh-500", None); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + let (_peer, outbound_req) = parse_sent_request(&mut rx); + let outbound_cseq = outbound_req + .header_value("CSeq") + .expect("outbound CSeq") + .to_string(); + + let resp = SipResponseBuilder::new(500, "Server Error") + .header("Call-ID", "call-refresh-500") + .header("CSeq", outbound_cseq) + .build(); + let events = core.handle_response(resp, dummy_peer()); + + assert!(events.is_empty(), "500 should not emit refresh events"); + assert!(core + .invites + .get(&call_id) + .expect("invite context should remain") + .pending_refresh + .is_none()); + + let (_peer, ack) = parse_sent_request(&mut rx); + match ack.method { + SipMethod::Ack => {} + other => panic!("expected ACK, got {:?}", other), + } + assert!( + rx.try_recv().is_err(), + "500 refresh failure must not send BYE immediately" + ); + } + #[test] fn rseq_random_and_increment() { let mut ctx = dummy_invite_context(); diff --git a/virtual-voicebot-backend/src/shared/ports/sip.rs b/virtual-voicebot-backend/src/shared/ports/sip.rs index 463ea7f..a944691 100644 --- a/virtual-voicebot-backend/src/shared/ports/sip.rs +++ b/virtual-voicebot-backend/src/shared/ports/sip.rs @@ -71,6 +71,10 @@ pub enum SipEvent { call_id: CallId, timer: SessionTimerInfo, }, + /// outbound session refresh が失敗し、通話終了へ移行するときの通知 + SessionRefreshFailed { + call_id: CallId, + }, Unknown, } @@ -85,8 +89,11 @@ pub enum SipCommand { Send183 { answer: Sdp }, /// SIP final (200 + SDP) Send200 { answer: Sdp }, - /// SIP UPDATE によるセッションリフレッシュ - SendUpdate { expires: Duration }, + /// SIP UPDATE / re-INVITE によるセッションリフレッシュ + SendSessionRefresh { + expires: Duration, + local_sdp: Option, + }, /// SIP エラー応答（INVITE の最終応答） SendError { code: u16, reason: String }, /// SIP BYE送信（UAC側として終話） From 6c587939e5861470d2b46a2290d0fea6d8da1aee Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Tue, 10 Mar 2026 21:23:42 +0900 Subject: [PATCH 2/7] =?UTF-8?q?#280=20=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/protocol/sip/core.rs | 54 +++++++++++++++---- 1 file changed, 43 insertions(+), 11 deletions(-) diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index 73ce20e..5713fac 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -267,6 +267,15 @@ fn parse_min_se(value: &str) -> Option { value.trim().parse::().ok() } +fn sync_local_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { + let remote_cseq = req + .header_value("CSeq") + .and_then(|value| parse_cseq_header(value).ok()) + .map(|cseq| cseq.num) + .unwrap_or(0); + ctx.local_cseq = ctx.local_cseq.max(remote_cseq); +} + const RSEQ_MIN: u32 = 1; const RSEQ_MAX: u32 = 0x7FFF_FFFF; @@ -1265,6 +1274,8 @@ impl SipCore { local_cseq: 0, expires_at: None, }; + let mut ctx = ctx; + sync_local_cseq_with_request(&mut ctx, &req); if is_outbound_invite { self.outbound_call_id = Some(headers.call_id.clone()); } @@ -1362,6 +1373,7 @@ impl SipCore { ctx.final_ok = None; ctx.final_ok_payload = None; ctx.pending_refresh = None; + sync_local_cseq_with_request(ctx, &req); if let Some(cfg) = &session_timer { ctx.session_timer = Some(cfg.clone()); } @@ -1689,13 +1701,14 @@ impl SipCore { } let bytes = resp.to_bytes(); tx.on_final_sent(bytes.clone()); - tx.last_request = Some(req); + tx.last_request = Some(req.clone()); self.send_payload(peer, bytes); } if let Some(cfg) = session_timer { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.session_timer = Some(cfg.clone()); + sync_local_cseq_with_request(ctx, &req); } vec![SipEvent::SessionRefresh { call_id: headers.call_id, @@ -2291,13 +2304,17 @@ mod tests { (core, rx) } - fn inbound_invite_request(call_id: &str, allow: Option<&str>) -> SipRequest { + fn inbound_invite_request_with_cseq( + call_id: &str, + allow: Option<&str>, + cseq: u32, + ) -> SipRequest { let mut builder = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com") .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") .header("From", ";tag=alice") .header("To", "") .header("Call-ID", call_id) - .header("CSeq", "1 INVITE"); + .header("CSeq", format!("{cseq} INVITE")); if let Some(allow) = allow { builder = builder.header("Allow", allow); } @@ -2305,7 +2322,16 @@ mod tests { } fn send_inbound_invite(core: &mut SipCore, call_id: &str, allow: Option<&str>) -> CallId { - let req = inbound_invite_request(call_id, allow); + send_inbound_invite_with_cseq(core, call_id, allow, 1) + } + + fn send_inbound_invite_with_cseq( + core: &mut SipCore, + call_id: &str, + allow: Option<&str>, + cseq: u32, + ) -> CallId { + let req = inbound_invite_request_with_cseq(call_id, allow, cseq); let input = SipInput { peer: dummy_peer(), data: req.to_bytes(), @@ -2370,7 +2396,7 @@ mod tests { #[test] fn session_refresh_uses_reinvite_when_update_not_allowed() { let (mut core, mut rx) = test_core(); - let call_id = send_inbound_invite(&mut core, "call-reinvite", None); + let call_id = send_inbound_invite_with_cseq(&mut core, "call-reinvite", None, 41); core.handle_sip_command( &call_id, @@ -2385,6 +2411,7 @@ mod tests { SipMethod::Invite => {} other => panic!("expected INVITE, got {:?}", other), } + assert_eq!(req.header_value("CSeq"), Some("42 INVITE")); assert_eq!( req.header_value("Session-Expires"), Some("90;refresher=uas") @@ -2403,8 +2430,12 @@ mod tests { #[test] fn session_refresh_uses_update_when_peer_supports_it() { let (mut core, mut rx) = test_core(); - let call_id = - send_inbound_invite(&mut core, "call-update", Some("INVITE, ACK, UPDATE, BYE")); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-update", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); core.handle_sip_command( &call_id, @@ -2419,6 +2450,7 @@ mod tests { SipMethod::Update => {} other => panic!("expected UPDATE, got {:?}", other), } + assert_eq!(req.header_value("CSeq"), Some("12 UPDATE")); assert_eq!( req.header_value("Session-Expires"), Some("120;refresher=uas") @@ -2436,7 +2468,7 @@ mod tests { #[test] fn reinvite_refresh_2xx_sends_ack_and_emits_session_refresh() { let (mut core, mut rx) = test_core(); - let call_id = send_inbound_invite(&mut core, "call-refresh-ok", None); + let call_id = send_inbound_invite_with_cseq(&mut core, "call-refresh-ok", None, 21); core.handle_sip_command( &call_id, @@ -2477,13 +2509,13 @@ mod tests { SipMethod::Ack => {} other => panic!("expected ACK, got {:?}", other), } - assert_eq!(ack.header_value("CSeq"), Some("1 ACK")); + assert_eq!(ack.header_value("CSeq"), Some("22 ACK")); } #[test] fn reinvite_refresh_408_sends_ack_then_bye_and_emits_failure() { let (mut core, mut rx) = test_core(); - let call_id = send_inbound_invite(&mut core, "call-refresh-408", None); + let call_id = send_inbound_invite_with_cseq(&mut core, "call-refresh-408", None, 31); core.handle_sip_command( &call_id, @@ -2525,7 +2557,7 @@ mod tests { #[test] fn reinvite_refresh_other_failure_does_not_send_bye() { let (mut core, mut rx) = test_core(); - let call_id = send_inbound_invite(&mut core, "call-refresh-500", None); + let call_id = send_inbound_invite_with_cseq(&mut core, "call-refresh-500", None, 51); core.handle_sip_command( &call_id, From f5424c75bbb8866888b3c7cefd41dcad5eb3598d Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Tue, 10 Mar 2026 22:06:01 +0900 Subject: [PATCH 3/7] =?UTF-8?q?#280=20coderabiit/codex=20bot=E6=8C=87?= =?UTF-8?q?=E6=91=98=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- virtual-voicebot-backend/src/main.rs | 15 +- .../src/protocol/session/handlers/mod.rs | 86 ++++++- .../src/protocol/session/state_machine.rs | 16 ++ .../src/protocol/session/types.rs | 1 + .../src/protocol/sip/core.rs | 230 ++++++++++++++---- 5 files changed, 294 insertions(+), 54 deletions(-) diff --git a/virtual-voicebot-backend/src/main.rs b/virtual-voicebot-backend/src/main.rs index 6d46df5..8ba4c9a 100644 --- a/virtual-voicebot-backend/src/main.rs +++ b/virtual-voicebot-backend/src/main.rs @@ -609,10 +609,21 @@ async fn main() -> anyhow::Result<()> { sip_core.handle_sip_command(&call_id, SipCommand::Send200 { answer }); } SessionOut::SipSendSessionRefresh { expires, local_sdp } => { - sip_core.handle_sip_command( + for event in sip_core.handle_sip_command( &call_id, SipCommand::SendSessionRefresh { expires, local_sdp }, - ); + ) { + if let SipEvent::SessionRefreshFailed { call_id } = event { + if let Some(sess_tx) = session_registry.get(&call_id).await { + let _ = sess_tx + .control_tx + .send(SessionControlIn::SipSessionRefreshFailed { + call_id: call_id.clone(), + }) + .await; + } + } + } } SessionOut::SipSendError { code, reason } => { sip_core.handle_sip_command( diff --git a/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs b/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs index 1e514c7..8c60b37 100644 --- a/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs +++ b/virtual-voicebot-backend/src/protocol/session/handlers/mod.rs @@ -684,13 +684,22 @@ impl SessionCoordinator { if let (Some(expires), Some(SessionRefresher::Uas)) = (self.session_expires, self.session_refresher) { - let _ = self.session_out_tx.try_send(( - self.call_id.clone(), - SessionOut::SipSendSessionRefresh { - expires, - local_sdp: self.local_sdp.clone(), - }, - )); + if let Err(err) = self + .session_out_tx + .send(( + self.call_id.clone(), + SessionOut::SipSendSessionRefresh { + expires, + local_sdp: self.local_sdp.clone(), + }, + )) + .await + { + warn!( + "[session {}] failed to emit session refresh request: {:?}", + self.call_id, err + ); + } } } (_, SessionControlIn::SessionTimerFired) => { @@ -1719,7 +1728,19 @@ mod tests { AppEventRx, mpsc::Receiver, ) { - let (session_out_tx, session_out_rx) = mpsc::channel(32); + build_test_session_with_session_out_capacity(routing_port, 32) + } + + fn build_test_session_with_session_out_capacity( + routing_port: Arc, + session_out_capacity: usize, + ) -> ( + SessionCoordinator, + mpsc::Receiver<(CallId, SessionOut)>, + AppEventRx, + mpsc::Receiver, + ) { + let (session_out_tx, session_out_rx) = mpsc::channel(session_out_capacity); let (app_tx, app_rx) = app_event_channel(16); let (control_tx, control_rx) = mpsc::channel(super::super::SESSION_CONTROL_CHANNEL_CAPACITY); @@ -2502,6 +2523,55 @@ mod tests { } } + #[tokio::test] + async fn session_refresh_due_waits_for_session_out_capacity_instead_of_dropping() { + let routing_port = Arc::new(NoopRoutingPort::new()); + let (mut session, mut session_out_rx, _app_rx, _control_rx) = + build_test_session_with_session_out_capacity(routing_port, 1); + session.local_sdp = Some(Sdp::pcmu("127.0.0.1", 10000)); + session.session_expires = Some(Duration::from_secs(90)); + session.session_refresher = Some(SessionRefresher::Uas); + + session + .session_out_tx + .send((session.call_id.clone(), SessionOut::RtpStopTx)) + .await + .expect("prefill session output channel"); + + let drain_task = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(20)).await; + let first = session_out_rx.recv().await.expect("prefill output"); + let second = timeout(Duration::from_millis(100), session_out_rx.recv()) + .await + .expect("session refresh output should arrive after capacity is freed") + .expect("session refresh output"); + (first, second) + }); + + let advance = timeout( + Duration::from_millis(200), + session + .handle_control_event(SessState::Established, SessionControlIn::SessionRefreshDue), + ) + .await + .expect("session refresh send should wait for channel capacity"); + + assert!(advance); + + let (first, second) = drain_task.await.expect("drain task should complete"); + assert!(matches!(first.1, SessionOut::RtpStopTx)); + match second.1 { + SessionOut::SipSendSessionRefresh { expires, local_sdp } => { + assert_eq!(expires, Duration::from_secs(90)); + assert!(matches!( + local_sdp, + Some(sdp) if sdp.ip == "127.0.0.1" && sdp.port == 10000 + )); + } + other => panic!("expected SipSendSessionRefresh, got {:?}", other), + } + } + #[tokio::test] async fn sip_session_refresh_failed_stops_timers_and_emits_timeout_end() { let routing_port = Arc::new(NoopRoutingPort::new()); diff --git a/virtual-voicebot-backend/src/protocol/session/state_machine.rs b/virtual-voicebot-backend/src/protocol/session/state_machine.rs index 8fdefa0..d71363b 100644 --- a/virtual-voicebot-backend/src/protocol/session/state_machine.rs +++ b/virtual-voicebot-backend/src/protocol/session/state_machine.rs @@ -75,4 +75,20 @@ mod tests { let commands = sm.process_event(SessionEvent::from(&event)); assert_eq!(commands, vec![SessionCommand::Transition(SessState::Early)]); } + + #[test] + fn process_event_terminates_on_sip_session_refresh_failed() { + let mut sm = SessionStateMachine::new(); + sm.apply_commands(&[SessionCommand::Transition(SessState::Established)]); + + let event = SessionControlIn::SipSessionRefreshFailed { + call_id: CallId::new("call".to_string()).expect("valid test call id"), + }; + + let commands = sm.process_event(SessionEvent::from(&event)); + assert_eq!( + commands, + vec![SessionCommand::Transition(SessState::Terminated)] + ); + } } diff --git a/virtual-voicebot-backend/src/protocol/session/types.rs b/virtual-voicebot-backend/src/protocol/session/types.rs index 5a9c5af..9f7fc01 100644 --- a/virtual-voicebot-backend/src/protocol/session/types.rs +++ b/virtual-voicebot-backend/src/protocol/session/types.rs @@ -287,6 +287,7 @@ pub(crate) fn next_session_state(current: SessState, event: &SessionControlIn) - | SessionControlIn::BLegBye | SessionControlIn::AppHangup | SessionControlIn::SessionTimerFired + | SessionControlIn::SipSessionRefreshFailed { .. } | SessionControlIn::Abort(_) => SessState::Terminated, SessionControlIn::RingDurationElapsed => current, SessionControlIn::SipInvite { .. } => match current { diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index 5713fac..571fa32 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -40,6 +40,7 @@ pub struct SipCore { struct InviteContext { tx: InviteServerTransaction, req: SipRequest, + remote_target_uri: String, reliable: Option, expected_rack: Option, last_rseq: Option, @@ -95,6 +96,7 @@ struct PendingSessionRefresh { cseq: u32, expires: Duration, local_sdp: Option, + via: Option, } #[derive(Debug, Clone)] @@ -232,6 +234,25 @@ fn response_header_value<'a>(resp: &'a SipResponse, name: &str) -> Option<&'a st .map(|header| header.value.as_str()) } +fn payload_header_value<'a>(payload: &'a [u8], name: &str) -> Option<&'a str> { + for raw_line in payload.split(|b| *b == b'\n') { + let Ok(line) = std::str::from_utf8(raw_line) else { + continue; + }; + let line = line.trim_end_matches('\r'); + if line.is_empty() { + break; + } + let Some((header_name, value)) = line.split_once(':') else { + continue; + }; + if header_name.eq_ignore_ascii_case(name) { + return Some(value.trim()); + } + } + None +} + fn supports_100rel(req: &SipRequest) -> bool { header_has_token(req.header_value("Supported"), "100rel") || header_has_token(req.header_value("Require"), "100rel") @@ -276,6 +297,19 @@ fn sync_local_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { ctx.local_cseq = ctx.local_cseq.max(remote_cseq); } +fn request_remote_target_uri(req: &SipRequest) -> String { + req.header_value("Contact") + .map(extract_contact_uri) + .unwrap_or_else(|| req.uri.as_str()) + .to_string() +} + +fn response_remote_target_uri(resp: &SipResponse) -> Option { + response_header_value(resp, "Contact") + .map(extract_contact_uri) + .map(str::to_string) +} + const RSEQ_MIN: u32 = 1; const RSEQ_MAX: u32 = 0x7FFF_FFFF; @@ -480,11 +514,7 @@ fn build_session_refresh_request( .header_value("Call-ID") .ok_or(SessionRefreshBuildError::MissingHeaders)? .to_string(); - let uri = req - .header_value("Contact") - .map(extract_contact_uri) - .unwrap_or_else(|| req.uri.as_str()) - .to_string(); + let uri = ctx.remote_target_uri.clone(); let req_uri = req.uri.clone(); let transport = match ctx.tx.peer { TransportPeer::Udp(_) => "UDP", @@ -531,6 +561,15 @@ fn build_session_refresh_request( } fn build_ack_request(ctx: &InviteContext, cfg: &SipConfig, invite_cseq: u32) -> Option> { + build_ack_request_with_via(ctx, cfg, invite_cseq, None) +} + +fn build_ack_request_with_via( + ctx: &InviteContext, + cfg: &SipConfig, + invite_cseq: u32, + via_override: Option<&str>, +) -> Option> { let req = &ctx.req; let to = req.header_value("From")?.to_string(); let mut from = req.header_value("To")?.to_string(); @@ -538,22 +577,20 @@ fn build_ack_request(ctx: &InviteContext, cfg: &SipConfig, invite_cseq: u32) -> from = format!("{from};tag=rustbot"); } let call_id = req.header_value("Call-ID")?.to_string(); - let uri = req - .header_value("Contact") - .map(extract_contact_uri) - .unwrap_or_else(|| req.uri.as_str()) - .to_string(); + let uri = ctx.remote_target_uri.clone(); let transport = match ctx.tx.peer { TransportPeer::Udp(_) => "UDP", TransportPeer::Tcp(_) => "TCP", }; - let via = format!( - "SIP/2.0/{} {}:{};branch={}", - transport, - cfg.advertised_ip, - cfg.sip_port, - generate_branch() - ); + let via = via_override.map(str::to_string).unwrap_or_else(|| { + format!( + "SIP/2.0/{} {}:{};branch={}", + transport, + cfg.advertised_ip, + cfg.sip_port, + generate_branch() + ) + }); Some( SipRequestBuilder::new(SipMethod::Ack, uri) .header("Via", via) @@ -575,11 +612,7 @@ fn build_bye_request(ctx: &mut InviteContext, cfg: &SipConfig) -> Option from = format!("{from};tag=rustbot"); } let call_id = req.header_value("Call-ID")?.to_string(); - let uri = req - .header_value("Contact") - .map(extract_contact_uri) - .unwrap_or_else(|| req.uri.as_str()) - .to_string(); + let uri = ctx.remote_target_uri.clone(); let transport = match ctx.tx.peer { TransportPeer::Udp(_) => "UDP", TransportPeer::Tcp(_) => "TCP", @@ -988,7 +1021,7 @@ impl SipCore { if let Some(call_id) = response_header_value(&resp, "Call-ID") .and_then(|value| CallId::new(value.to_string()).ok()) { - if let Some(action) = self.handle_session_refresh_response(&call_id, &resp) { + if let Some(action) = self.handle_session_refresh_response(&call_id, &resp, peer) { if let Some((peer, payload)) = action.ack_payload { self.send_payload(peer, payload); } @@ -1023,6 +1056,7 @@ impl SipCore { &mut self, call_id: &CallId, resp: &SipResponse, + response_peer: TransportPeer, ) -> Option { let cseq = response_header_value(resp, "CSeq").and_then(|value| parse_cseq_header(value).ok())?; @@ -1036,7 +1070,6 @@ impl SipCore { return None; } - let peer = ctx.tx.peer; let mut action = SessionRefreshResponseAction { call_id: call_id.clone(), ack_payload: None, @@ -1048,13 +1081,17 @@ impl SipCore { if resp.status_code < 200 { return Some(action); } - if pending.method == SessionRefreshMethod::Invite && resp.status_code >= 200 { - action.ack_payload = - build_ack_request(ctx, &self.cfg, pending.cseq).map(|payload| (peer, payload)); - } match resp.status_code { 200..=299 => { + ctx.tx.peer = response_peer; + if let Some(remote_target_uri) = response_remote_target_uri(resp) { + ctx.remote_target_uri = remote_target_uri; + } + if pending.method == SessionRefreshMethod::Invite { + action.ack_payload = build_ack_request(ctx, &self.cfg, pending.cseq) + .map(|payload| (response_peer, payload)); + } ctx.pending_refresh = None; if let Some(timer) = session_timer_info_from_response(resp, ctx.session_timer.as_ref()) @@ -1075,6 +1112,15 @@ impl SipCore { } } 422 => { + if pending.method == SessionRefreshMethod::Invite { + action.ack_payload = build_ack_request_with_via( + ctx, + &self.cfg, + pending.cseq, + pending.via.as_deref(), + ) + .map(|payload| (response_peer, payload)); + } let Some(min_se) = response_header_value(resp, "Min-SE").and_then(parse_min_se) else { ctx.pending_refresh = None; @@ -1108,8 +1154,9 @@ impl SipCore { cseq: ctx.local_cseq, expires: retry_expires, local_sdp: pending.local_sdp.clone(), + via: payload_header_value(&payload, "Via").map(str::to_string), }); - action.retry_payload = Some((peer, payload)); + action.retry_payload = Some((ctx.tx.peer, payload)); } Err(SessionRefreshBuildError::MissingLocalSdp) => { ctx.pending_refresh = None; @@ -1128,7 +1175,7 @@ impl SipCore { Err(SessionRefreshBuildError::CseqOverflow) => { ctx.pending_refresh = None; action.bye_payload = - build_bye_request(ctx, &self.cfg).map(|payload| (peer, payload)); + build_bye_request(ctx, &self.cfg).map(|payload| (ctx.tx.peer, payload)); action.events.push(SipEvent::SessionRefreshFailed { call_id: call_id.clone(), }); @@ -1137,15 +1184,33 @@ impl SipCore { } } 408 | 481 => { + if pending.method == SessionRefreshMethod::Invite { + action.ack_payload = build_ack_request_with_via( + ctx, + &self.cfg, + pending.cseq, + pending.via.as_deref(), + ) + .map(|payload| (response_peer, payload)); + } ctx.pending_refresh = None; action.bye_payload = - build_bye_request(ctx, &self.cfg).map(|payload| (peer, payload)); + build_bye_request(ctx, &self.cfg).map(|payload| (ctx.tx.peer, payload)); action.events.push(SipEvent::SessionRefreshFailed { call_id: call_id.clone(), }); action.remove_invite = true; } _ => { + if pending.method == SessionRefreshMethod::Invite { + action.ack_payload = build_ack_request_with_via( + ctx, + &self.cfg, + pending.cseq, + pending.via.as_deref(), + ) + .map(|payload| (response_peer, payload)); + } ctx.pending_refresh = None; } } @@ -1263,6 +1328,7 @@ impl SipCore { let ctx = InviteContext { tx, req: req.clone(), + remote_target_uri: request_remote_target_uri(&req), reliable: None, expected_rack: None, last_rseq: None, @@ -1276,6 +1342,7 @@ impl SipCore { }; let mut ctx = ctx; sync_local_cseq_with_request(&mut ctx, &req); + ctx.remote_target_uri = request_remote_target_uri(&req); if is_outbound_invite { self.outbound_call_id = Some(headers.call_id.clone()); } @@ -1365,6 +1432,7 @@ impl SipCore { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.req = req.clone(); + ctx.remote_target_uri = request_remote_target_uri(&req); ctx.tx = InviteServerTransaction::new(peer); ctx.tx.invite_req = Some(req.clone()); ctx.reliable = None; @@ -1708,6 +1776,8 @@ impl SipCore { if let Some(cfg) = session_timer { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.session_timer = Some(cfg.clone()); + ctx.remote_target_uri = request_remote_target_uri(&req); + ctx.tx.peer = peer; sync_local_cseq_with_request(ctx, &req); } vec![SipEvent::SessionRefresh { @@ -1898,7 +1968,8 @@ impl SipCore { /// core.handle_sip_command(call_id, cmd); /// # } /// ``` - pub fn handle_sip_command(&mut self, call_id: &CallId, cmd: SipCommand) { + pub fn handle_sip_command(&mut self, call_id: &CallId, cmd: SipCommand) -> Vec { + let mut events = Vec::new(); match cmd { SipCommand::Send100 => { if let Some(ctx) = self.invites.get_mut(call_id) { @@ -2056,16 +2127,12 @@ impl SipCore { SessionRefreshMethod::Update => { build_update_request(ctx, &self.cfg, expires) } - SessionRefreshMethod::Invite => { - let Some(local_sdp) = local_sdp.as_ref() else { - log::warn!( - "[sip refresh] local SDP missing, cannot send re-INVITE call_id={}", - call_id - ); - return; - }; - build_reinvite_request(ctx, &self.cfg, local_sdp, expires) - } + SessionRefreshMethod::Invite => match local_sdp.as_ref() { + Some(local_sdp) => { + build_reinvite_request(ctx, &self.cfg, local_sdp, expires) + } + None => Err(SessionRefreshBuildError::MissingLocalSdp), + }, }; Some((peer, method, payload)) } else { @@ -2080,6 +2147,7 @@ impl SipCore { cseq: ctx.local_cseq, expires, local_sdp, + via: payload_header_value(&payload, "Via").map(str::to_string), }); } self.send_payload(peer, payload); @@ -2097,6 +2165,24 @@ impl SipCore { method, call_id ); + if method == SessionRefreshMethod::Invite { + let bye = if let Some(ctx) = self.invites.get_mut(call_id) { + build_bye_request(ctx, &self.cfg) + .map(|payload| (ctx.tx.peer, payload)) + } else { + None + }; + if let Some((peer, payload)) = bye { + self.send_payload(peer, payload); + } + if self.outbound_call_id.as_ref() == Some(call_id) { + self.outbound_call_id = None; + } + self.invites.remove(call_id); + events.push(SipEvent::SessionRefreshFailed { + call_id: call_id.clone(), + }); + } } Err(SessionRefreshBuildError::CseqOverflow) => { log::warn!( @@ -2180,6 +2266,7 @@ impl SipCore { self.invites.remove(call_id); } } + events } fn extract_call_id_from_payload(payload: &[u8]) -> Option<&str> { @@ -2291,6 +2378,10 @@ mod tests { TransportPeer::Udp("127.0.0.1:5060".parse().unwrap()) } + fn refreshed_peer() -> TransportPeer { + TransportPeer::Udp("127.0.0.1:5090".parse().unwrap()) + } + fn test_core() -> (SipCore, mpsc::Receiver) { let (tx, rx) = mpsc::channel(16); let core = SipCore::new( @@ -2361,6 +2452,7 @@ mod tests { InviteContext { tx: InviteServerTransaction::new(dummy_peer()), req, + remote_target_uri: "sip:test@example.com".to_string(), reliable: None, expected_rack: None, last_rseq: None, @@ -2465,6 +2557,38 @@ mod tests { assert_eq!(pending.method, SessionRefreshMethod::Update); } + #[test] + fn reinvite_refresh_missing_local_sdp_sends_bye_and_emits_failure() { + let (mut core, mut rx) = test_core(); + let call_id = + send_inbound_invite_with_cseq(&mut core, "call-refresh-missing-sdp", None, 61); + + let events = core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: None, + }, + ); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefreshFailed { call_id }] + if call_id.as_str() == "call-refresh-missing-sdp" + )); + assert!(!core.invites.contains_key(&call_id)); + + let (_peer, bye) = parse_sent_request(&mut rx); + match bye.method { + SipMethod::Bye => {} + other => panic!("expected BYE, got {:?}", other), + } + assert!( + rx.try_recv().is_err(), + "missing local SDP should fail before sending a re-INVITE" + ); + } + #[test] fn reinvite_refresh_2xx_sends_ack_and_emits_session_refresh() { let (mut core, mut rx) = test_core(); @@ -2486,9 +2610,10 @@ mod tests { let resp = SipResponseBuilder::new(200, "OK") .header("Call-ID", "call-refresh-ok") .header("CSeq", outbound_cseq) + .header("Contact", "") .header("Session-Expires", "180;refresher=uas") .build(); - let events = core.handle_response(resp, dummy_peer()); + let events = core.handle_response(resp, refreshed_peer()); assert!(matches!( events.as_slice(), @@ -2504,12 +2629,19 @@ mod tests { .pending_refresh .is_none()); - let (_peer, ack) = parse_sent_request(&mut rx); + let (ack_peer, ack) = parse_sent_request(&mut rx); + assert_eq!(ack_peer, refreshed_peer()); match ack.method { SipMethod::Ack => {} other => panic!("expected ACK, got {:?}", other), } + assert_eq!(ack.uri, "sip:target-refresh@example.net"); assert_eq!(ack.header_value("CSeq"), Some("22 ACK")); + + core.handle_sip_command(&call_id, SipCommand::SendBye); + let (bye_peer, bye) = parse_sent_request(&mut rx); + assert_eq!(bye_peer, refreshed_peer()); + assert_eq!(bye.uri, "sip:target-refresh@example.net"); } #[test] @@ -2529,6 +2661,10 @@ mod tests { .header_value("CSeq") .expect("outbound CSeq") .to_string(); + let outbound_via = outbound_req + .header_value("Via") + .expect("outbound Via") + .to_string(); let resp = SipResponseBuilder::new(408, "Request Timeout") .header("Call-ID", "call-refresh-408") @@ -2547,6 +2683,7 @@ mod tests { SipMethod::Ack => {} other => panic!("expected ACK, got {:?}", other), } + assert_eq!(ack.header_value("Via"), Some(outbound_via.as_str())); let (_peer, bye) = parse_sent_request(&mut rx); match bye.method { SipMethod::Bye => {} @@ -2571,6 +2708,10 @@ mod tests { .header_value("CSeq") .expect("outbound CSeq") .to_string(); + let outbound_via = outbound_req + .header_value("Via") + .expect("outbound Via") + .to_string(); let resp = SipResponseBuilder::new(500, "Server Error") .header("Call-ID", "call-refresh-500") @@ -2591,6 +2732,7 @@ mod tests { SipMethod::Ack => {} other => panic!("expected ACK, got {:?}", other), } + assert_eq!(ack.header_value("Via"), Some(outbound_via.as_str())); assert!( rx.try_recv().is_err(), "500 refresh failure must not send BYE immediately" From 91ebd7c1567196fa7b4bbf1b1f4717b9b0464366 Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Tue, 10 Mar 2026 22:32:11 +0900 Subject: [PATCH 4/7] =?UTF-8?q?#280=20coderabiit=E6=8C=87=E6=91=98?= =?UTF-8?q?=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- virtual-voicebot-backend/src/main.rs | 131 ++++++++++++++---- .../src/protocol/sip/core.rs | 93 +++++++++++-- 2 files changed, 189 insertions(+), 35 deletions(-) diff --git a/virtual-voicebot-backend/src/main.rs b/virtual-voicebot-backend/src/main.rs index 8ba4c9a..605216c 100644 --- a/virtual-voicebot-backend/src/main.rs +++ b/virtual-voicebot-backend/src/main.rs @@ -72,6 +72,25 @@ fn sanitize_optional_url(url: Option<&str>) -> String { .unwrap_or_else(|| "none".to_string()) } +async fn forward_refresh_failure(session_registry: &SessionRegistry, call_id: &CallId) { + if let Some(sess_tx) = session_registry.get(call_id).await { + let _ = sess_tx + .control_tx + .send(SessionControlIn::SipSessionRefreshFailed { + call_id: call_id.clone(), + }) + .await; + } +} + +async fn forward_sip_command_events(session_registry: &SessionRegistry, events: Vec) { + for event in events { + if let SipEvent::SessionRefreshFailed { call_id } = event { + forward_refresh_failure(session_registry, &call_id).await; + } + } +} + fn log_ai_config() { let ai = config::ai_config(); @@ -493,14 +512,7 @@ async fn main() -> anyhow::Result<()> { } } SipEvent::SessionRefreshFailed { call_id } => { - if let Some(sess_tx) = session_registry.get(&call_id).await { - let _ = sess_tx - .control_tx - .send(SessionControlIn::SipSessionRefreshFailed { - call_id: call_id.clone(), - }) - .await; - } + forward_refresh_failure(&session_registry, &call_id).await; } SipEvent::TransactionTimeout { call_id } => { log::warn!("[main] TransactionTimeout for call_id={}", call_id); @@ -609,21 +621,14 @@ async fn main() -> anyhow::Result<()> { sip_core.handle_sip_command(&call_id, SipCommand::Send200 { answer }); } SessionOut::SipSendSessionRefresh { expires, local_sdp } => { - for event in sip_core.handle_sip_command( - &call_id, - SipCommand::SendSessionRefresh { expires, local_sdp }, - ) { - if let SipEvent::SessionRefreshFailed { call_id } = event { - if let Some(sess_tx) = session_registry.get(&call_id).await { - let _ = sess_tx - .control_tx - .send(SessionControlIn::SipSessionRefreshFailed { - call_id: call_id.clone(), - }) - .await; - } - } - } + forward_sip_command_events( + &session_registry, + sip_core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { expires, local_sdp }, + ), + ) + .await; } SessionOut::SipSendError { code, reason } => { sip_core.handle_sip_command( @@ -648,7 +653,41 @@ async fn main() -> anyhow::Result<()> { #[cfg(test)] mod tests { - use super::{sanitize_optional_url, sanitized_display_url_for_log}; + use super::{ + forward_refresh_failure, forward_sip_command_events, sanitize_optional_url, + sanitized_display_url_for_log, + }; + use tokio::sync::mpsc; + use tokio::time::{timeout, Duration}; + use virtual_voicebot_backend::protocol::session::types::CallId; + use virtual_voicebot_backend::protocol::session::{ + SessionControlIn, SessionHandle, SessionMediaIn, SessionRegistry, + }; + use virtual_voicebot_backend::protocol::sip::SipEvent; + + async fn build_test_session_registry( + call_id: &str, + ) -> ( + SessionRegistry, + CallId, + mpsc::Receiver, + mpsc::Receiver, + ) { + let registry = SessionRegistry::new(); + let call_id = CallId::new(call_id.to_string()).expect("valid test call id"); + let (control_tx, control_rx) = mpsc::channel(1); + let (media_tx, media_rx) = mpsc::channel(1); + registry + .insert( + call_id.clone(), + SessionHandle { + control_tx, + media_tx, + }, + ) + .await; + (registry, call_id, control_rx, media_rx) + } #[test] fn sanitized_display_url_for_log_returns_none_for_empty_or_whitespace() { @@ -703,4 +742,48 @@ mod tests { assert_eq!(sanitize_optional_url(Some(" ")), "none"); assert_eq!(sanitize_optional_url(None), "none"); } + + #[tokio::test] + async fn forward_refresh_failure_routes_to_session_control() { + let (registry, call_id, mut control_rx, _media_rx) = + build_test_session_registry("call-refresh-forward").await; + + forward_refresh_failure(®istry, &call_id).await; + + let control = timeout(Duration::from_millis(50), control_rx.recv()) + .await + .expect("refresh failure should be forwarded") + .expect("control event"); + assert!(matches!( + control, + SessionControlIn::SipSessionRefreshFailed { + call_id: forwarded_call_id, + } if forwarded_call_id == call_id + )); + } + + #[tokio::test] + async fn forward_sip_command_events_routes_immediate_refresh_failure() { + let (registry, call_id, mut control_rx, _media_rx) = + build_test_session_registry("call-refresh-command").await; + + forward_sip_command_events( + ®istry, + vec![SipEvent::SessionRefreshFailed { + call_id: call_id.clone(), + }], + ) + .await; + + let control = timeout(Duration::from_millis(50), control_rx.recv()) + .await + .expect("immediate refresh failure should be forwarded") + .expect("control event"); + assert!(matches!( + control, + SessionControlIn::SipSessionRefreshFailed { + call_id: forwarded_call_id, + } if forwarded_call_id == call_id + )); + } } diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index 571fa32..a25392d 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -1174,8 +1174,6 @@ impl SipCore { } Err(SessionRefreshBuildError::CseqOverflow) => { ctx.pending_refresh = None; - action.bye_payload = - build_bye_request(ctx, &self.cfg).map(|payload| (ctx.tx.peer, payload)); action.events.push(SipEvent::SessionRefreshFailed { call_id: call_id.clone(), }); @@ -2190,19 +2188,13 @@ impl SipCore { method, call_id ); - let bye = if let Some(ctx) = self.invites.get_mut(call_id) { - build_bye_request(ctx, &self.cfg) - .map(|payload| (ctx.tx.peer, payload)) - } else { - None - }; - if let Some((peer, payload)) = bye { - self.send_payload(peer, payload); - } if self.outbound_call_id.as_ref() == Some(call_id) { self.outbound_call_id = None; } self.invites.remove(call_id); + events.push(SipEvent::SessionRefreshFailed { + call_id: call_id.clone(), + }); } } } else { @@ -2589,6 +2581,32 @@ mod tests { ); } + #[test] + fn session_refresh_build_cseq_overflow_emits_failure_without_bye() { + let (mut core, mut rx) = test_core(); + let call_id = + send_inbound_invite_with_cseq(&mut core, "call-refresh-overflow", None, u32::MAX); + + let events = core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefreshFailed { call_id }] + if call_id.as_str() == "call-refresh-overflow" + )); + assert!(!core.invites.contains_key(&call_id)); + assert!( + rx.try_recv().is_err(), + "CSeq overflow should not send refresh requests or BYE" + ); + } + #[test] fn reinvite_refresh_2xx_sends_ack_and_emits_session_refresh() { let (mut core, mut rx) = test_core(); @@ -2691,6 +2709,59 @@ mod tests { } } + #[test] + fn reinvite_refresh_422_retry_cseq_overflow_emits_failure_without_bye() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-refresh-422-overflow", + None, + u32::MAX - 1, + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + let (_peer, outbound_req) = parse_sent_request(&mut rx); + let outbound_cseq = outbound_req + .header_value("CSeq") + .expect("outbound CSeq") + .to_string(); + let outbound_via = outbound_req + .header_value("Via") + .expect("outbound Via") + .to_string(); + + let resp = SipResponseBuilder::new(422, "Session Interval Too Small") + .header("Call-ID", "call-refresh-422-overflow") + .header("CSeq", outbound_cseq) + .header("Min-SE", "120") + .build(); + let events = core.handle_response(resp, dummy_peer()); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefreshFailed { call_id }] + if call_id.as_str() == "call-refresh-422-overflow" + )); + assert!(!core.invites.contains_key(&call_id)); + + let (_peer, ack) = parse_sent_request(&mut rx); + match ack.method { + SipMethod::Ack => {} + other => panic!("expected ACK, got {:?}", other), + } + assert_eq!(ack.header_value("Via"), Some(outbound_via.as_str())); + assert!( + rx.try_recv().is_err(), + "422 retry CSeq overflow should not send BYE after ACK" + ); + } + #[test] fn reinvite_refresh_other_failure_does_not_send_bye() { let (mut core, mut rx) = test_core(); From 0c4c74d0400def872ae323b177c7af7ac43ad002 Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Tue, 10 Mar 2026 23:02:27 +0900 Subject: [PATCH 5/7] =?UTF-8?q?#280=20coderabiit=E6=8C=87=E6=91=98?= =?UTF-8?q?=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/protocol/sip/core.rs | 238 ++++++++++++++++-- 1 file changed, 214 insertions(+), 24 deletions(-) diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index a25392d..5ce7a3b 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -41,6 +41,7 @@ struct InviteContext { tx: InviteServerTransaction, req: SipRequest, remote_target_uri: String, + remote_cseq: Option, reliable: Option, expected_rack: Option, last_rseq: Option, @@ -227,6 +228,10 @@ fn header_has_token(value: Option<&str>, token: &str) -> bool { .any(|part| part.eq_ignore_ascii_case(token)) } +fn request_allows_update(req: &SipRequest) -> bool { + header_has_token(req.header_value("Allow"), "UPDATE") +} + fn response_header_value<'a>(resp: &'a SipResponse, name: &str) -> Option<&'a str> { resp.headers .iter() @@ -234,6 +239,10 @@ fn response_header_value<'a>(resp: &'a SipResponse, name: &str) -> Option<&'a st .map(|header| header.value.as_str()) } +fn response_allows_update(resp: &SipResponse) -> bool { + header_has_token(response_header_value(resp, "Allow"), "UPDATE") +} + fn payload_header_value<'a>(payload: &'a [u8], name: &str) -> Option<&'a str> { for raw_line in payload.split(|b| *b == b'\n') { let Ok(line) = std::str::from_utf8(raw_line) else { @@ -288,13 +297,14 @@ fn parse_min_se(value: &str) -> Option { value.trim().parse::().ok() } -fn sync_local_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { - let remote_cseq = req +fn sync_remote_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { + if let Some(remote_cseq) = req .header_value("CSeq") .and_then(|value| parse_cseq_header(value).ok()) .map(|cseq| cseq.num) - .unwrap_or(0); - ctx.local_cseq = ctx.local_cseq.max(remote_cseq); + { + ctx.remote_cseq = Some(remote_cseq); + } } fn request_remote_target_uri(req: &SipRequest) -> String { @@ -1085,6 +1095,7 @@ impl SipCore { match resp.status_code { 200..=299 => { ctx.tx.peer = response_peer; + ctx.allow_update = response_allows_update(resp); if let Some(remote_target_uri) = response_remote_target_uri(resp) { ctx.remote_target_uri = remote_target_uri; } @@ -1247,11 +1258,7 @@ impl SipCore { .header_value("CSeq") .and_then(|value| parse_cseq_header(value).ok()) .map(|cseq| cseq.num); - let prev_cseq = ctx - .req - .header_value("CSeq") - .and_then(|value| parse_cseq_header(value).ok()) - .map(|cseq| cseq.num); + let prev_cseq = ctx.remote_cseq; if req_cseq.is_some() && prev_cseq.is_some() && req_cseq == prev_cseq { let (action, final_ok, tx_peer) = ( ctx.tx.on_retransmit(), @@ -1327,19 +1334,20 @@ impl SipCore { tx, req: req.clone(), remote_target_uri: request_remote_target_uri(&req), + remote_cseq: None, reliable: None, expected_rack: None, last_rseq: None, session_timer: session_timer.clone(), final_ok: None, final_ok_payload: None, - allow_update: header_has_token(req.header_value("Allow"), "UPDATE"), + allow_update: request_allows_update(&req), pending_refresh: None, local_cseq: 0, expires_at: None, }; let mut ctx = ctx; - sync_local_cseq_with_request(&mut ctx, &req); + sync_remote_cseq_with_request(&mut ctx, &req); ctx.remote_target_uri = request_remote_target_uri(&req); if is_outbound_invite { self.outbound_call_id = Some(headers.call_id.clone()); @@ -1431,6 +1439,7 @@ impl SipCore { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.req = req.clone(); ctx.remote_target_uri = request_remote_target_uri(&req); + ctx.allow_update = request_allows_update(&req); ctx.tx = InviteServerTransaction::new(peer); ctx.tx.invite_req = Some(req.clone()); ctx.reliable = None; @@ -1439,7 +1448,7 @@ impl SipCore { ctx.final_ok = None; ctx.final_ok_payload = None; ctx.pending_refresh = None; - sync_local_cseq_with_request(ctx, &req); + sync_remote_cseq_with_request(ctx, &req); if let Some(cfg) = &session_timer { ctx.session_timer = Some(cfg.clone()); } @@ -1775,8 +1784,9 @@ impl SipCore { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.session_timer = Some(cfg.clone()); ctx.remote_target_uri = request_remote_target_uri(&req); + ctx.allow_update = request_allows_update(&req); ctx.tx.peer = peer; - sync_local_cseq_with_request(ctx, &req); + sync_remote_cseq_with_request(ctx, &req); } vec![SipEvent::SessionRefresh { call_id: headers.call_id, @@ -2439,12 +2449,25 @@ mod tests { (sent.peer, req) } + fn parse_sent_response( + rx: &mut mpsc::Receiver, + ) -> (TransportPeer, SipResponse) { + let sent = rx.try_recv().expect("sent response"); + let text = String::from_utf8(sent.payload).expect("utf8 response"); + let resp = match parse_sip_message(&text).expect("parse response") { + SipMessage::Response(resp) => resp, + other => panic!("expected response, got {:?}", other), + }; + (sent.peer, resp) + } + fn dummy_invite_context() -> InviteContext { let req = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com").build(); InviteContext { tx: InviteServerTransaction::new(dummy_peer()), req, remote_target_uri: "sip:test@example.com".to_string(), + remote_cseq: None, reliable: None, expected_rack: None, last_rseq: None, @@ -2495,7 +2518,7 @@ mod tests { SipMethod::Invite => {} other => panic!("expected INVITE, got {:?}", other), } - assert_eq!(req.header_value("CSeq"), Some("42 INVITE")); + assert_eq!(req.header_value("CSeq"), Some("1 INVITE")); assert_eq!( req.header_value("Session-Expires"), Some("90;refresher=uas") @@ -2534,7 +2557,7 @@ mod tests { SipMethod::Update => {} other => panic!("expected UPDATE, got {:?}", other), } - assert_eq!(req.header_value("CSeq"), Some("12 UPDATE")); + assert_eq!(req.header_value("CSeq"), Some("1 UPDATE")); assert_eq!( req.header_value("Session-Expires"), Some("120;refresher=uas") @@ -2549,6 +2572,76 @@ mod tests { assert_eq!(pending.method, SessionRefreshMethod::Update); } + #[test] + fn session_refresh_falls_back_to_reinvite_after_response_drops_update_support() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-update-fallback", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + + let (_peer, first_req) = parse_sent_request(&mut rx); + match first_req.method { + SipMethod::Update => {} + other => panic!("expected UPDATE, got {:?}", other), + } + + let resp = SipResponseBuilder::new(200, "OK") + .header("Call-ID", "call-update-fallback") + .header( + "CSeq", + first_req + .header_value("CSeq") + .expect("outbound update CSeq"), + ) + .header("Contact", "") + .header("Allow", "INVITE, ACK, BYE") + .header("Session-Expires", "180;refresher=uas") + .build(); + let events = core.handle_response(resp, refreshed_peer()); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-update-fallback" + && timer.expires == Duration::from_secs(180) + && timer.refresher == SessionRefresher::Uas + )); + let ctx = core.invites.get(&call_id).expect("invite context"); + assert!( + !ctx.allow_update, + "Allow response without UPDATE should disable UPDATE refresh" + ); + assert_eq!(ctx.remote_target_uri, "sip:target-no-update@example.net"); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + + let (peer, second_req) = parse_sent_request(&mut rx); + assert_eq!(peer, refreshed_peer()); + match second_req.method { + SipMethod::Invite => {} + other => panic!("expected INVITE fallback, got {:?}", other), + } + assert_eq!(second_req.uri, "sip:target-no-update@example.net"); + assert_eq!(second_req.header_value("CSeq"), Some("2 INVITE")); + } + #[test] fn reinvite_refresh_missing_local_sdp_sends_bye_and_emits_failure() { let (mut core, mut rx) = test_core(); @@ -2584,8 +2677,11 @@ mod tests { #[test] fn session_refresh_build_cseq_overflow_emits_failure_without_bye() { let (mut core, mut rx) = test_core(); - let call_id = - send_inbound_invite_with_cseq(&mut core, "call-refresh-overflow", None, u32::MAX); + let call_id = send_inbound_invite_with_cseq(&mut core, "call-refresh-overflow", None, 41); + core.invites + .get_mut(&call_id) + .expect("invite context") + .local_cseq = u32::MAX; let events = core.handle_sip_command( &call_id, @@ -2607,6 +2703,100 @@ mod tests { ); } + #[test] + fn inbound_reinvite_does_not_advance_local_cseq() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq(&mut core, "call-remote-reinvite", None, 41); + + let reinvite = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-reinvite") + .header("CSeq", "99 INVITE") + .build(); + let events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: reinvite.to_bytes(), + }); + + assert!(matches!( + events.as_slice(), + [SipEvent::ReInvite { call_id, .. }] if call_id.as_str() == "call-remote-reinvite" + )); + assert_eq!( + core.invites + .get(&call_id) + .expect("invite context") + .remote_cseq, + Some(99) + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + + let (_peer, req) = parse_sent_request(&mut rx); + assert_eq!(req.header_value("CSeq"), Some("1 INVITE")); + } + + #[test] + fn inbound_update_does_not_advance_local_cseq() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-update", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + let update = SipRequestBuilder::new(SipMethod::Update, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-update") + .header("CSeq", "77 UPDATE") + .header("Allow", "INVITE, ACK, UPDATE, BYE") + .header("Contact", "") + .header("Session-Expires", "180;refresher=uas") + .build(); + let events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: update.to_bytes(), + }); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-remote-update" + && timer.expires == Duration::from_secs(180) + )); + let (_peer, resp) = parse_sent_response(&mut rx); + assert_eq!(resp.status_code, 200); + assert_eq!( + core.invites + .get(&call_id) + .expect("invite context") + .remote_cseq, + Some(77) + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: None, + }, + ); + + let (_peer, req) = parse_sent_request(&mut rx); + assert_eq!(req.header_value("CSeq"), Some("1 UPDATE")); + } + #[test] fn reinvite_refresh_2xx_sends_ack_and_emits_session_refresh() { let (mut core, mut rx) = test_core(); @@ -2654,7 +2844,7 @@ mod tests { other => panic!("expected ACK, got {:?}", other), } assert_eq!(ack.uri, "sip:target-refresh@example.net"); - assert_eq!(ack.header_value("CSeq"), Some("22 ACK")); + assert_eq!(ack.header_value("CSeq"), Some("1 ACK")); core.handle_sip_command(&call_id, SipCommand::SendBye); let (bye_peer, bye) = parse_sent_request(&mut rx); @@ -2712,12 +2902,12 @@ mod tests { #[test] fn reinvite_refresh_422_retry_cseq_overflow_emits_failure_without_bye() { let (mut core, mut rx) = test_core(); - let call_id = send_inbound_invite_with_cseq( - &mut core, - "call-refresh-422-overflow", - None, - u32::MAX - 1, - ); + let call_id = + send_inbound_invite_with_cseq(&mut core, "call-refresh-422-overflow", None, 41); + core.invites + .get_mut(&call_id) + .expect("invite context") + .local_cseq = u32::MAX - 1; core.handle_sip_command( &call_id, From e90eeaf185f9888722759fcdf5c60193a3eb9911 Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Wed, 11 Mar 2026 05:31:51 +0900 Subject: [PATCH 6/7] =?UTF-8?q?#280=20coderabiit=E6=8C=87=E6=91=98?= =?UTF-8?q?=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/protocol/sip/core.rs | 257 +++++++++++++++++- 1 file changed, 248 insertions(+), 9 deletions(-) diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index 5ce7a3b..c7b1d6b 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -228,8 +228,12 @@ fn header_has_token(value: Option<&str>, token: &str) -> bool { .any(|part| part.eq_ignore_ascii_case(token)) } -fn request_allows_update(req: &SipRequest) -> bool { - header_has_token(req.header_value("Allow"), "UPDATE") +fn allow_header_supports_update(value: Option<&str>) -> Option { + value.map(|allow| header_has_token(Some(allow), "UPDATE")) +} + +fn request_allows_update(req: &SipRequest) -> Option { + allow_header_supports_update(req.header_value("Allow")) } fn response_header_value<'a>(resp: &'a SipResponse, name: &str) -> Option<&'a str> { @@ -239,8 +243,8 @@ fn response_header_value<'a>(resp: &'a SipResponse, name: &str) -> Option<&'a st .map(|header| header.value.as_str()) } -fn response_allows_update(resp: &SipResponse) -> bool { - header_has_token(response_header_value(resp, "Allow"), "UPDATE") +fn response_allows_update(resp: &SipResponse) -> Option { + allow_header_supports_update(response_header_value(resp, "Allow")) } fn payload_header_value<'a>(payload: &'a [u8], name: &str) -> Option<&'a str> { @@ -303,7 +307,10 @@ fn sync_remote_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { .and_then(|value| parse_cseq_header(value).ok()) .map(|cseq| cseq.num) { - ctx.remote_cseq = Some(remote_cseq); + ctx.remote_cseq = Some( + ctx.remote_cseq + .map_or(remote_cseq, |existing| existing.max(remote_cseq)), + ); } } @@ -1095,7 +1102,9 @@ impl SipCore { match resp.status_code { 200..=299 => { ctx.tx.peer = response_peer; - ctx.allow_update = response_allows_update(resp); + if let Some(allow_update) = response_allows_update(resp) { + ctx.allow_update = allow_update; + } if let Some(remote_target_uri) = response_remote_target_uri(resp) { ctx.remote_target_uri = remote_target_uri; } @@ -1341,7 +1350,7 @@ impl SipCore { session_timer: session_timer.clone(), final_ok: None, final_ok_payload: None, - allow_update: request_allows_update(&req), + allow_update: request_allows_update(&req).unwrap_or(false), pending_refresh: None, local_cseq: 0, expires_at: None, @@ -1439,7 +1448,9 @@ impl SipCore { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.req = req.clone(); ctx.remote_target_uri = request_remote_target_uri(&req); - ctx.allow_update = request_allows_update(&req); + if let Some(allow_update) = request_allows_update(&req) { + ctx.allow_update = allow_update; + } ctx.tx = InviteServerTransaction::new(peer); ctx.tx.invite_req = Some(req.clone()); ctx.reliable = None; @@ -1784,7 +1795,9 @@ impl SipCore { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.session_timer = Some(cfg.clone()); ctx.remote_target_uri = request_remote_target_uri(&req); - ctx.allow_update = request_allows_update(&req); + if let Some(allow_update) = request_allows_update(&req) { + ctx.allow_update = allow_update; + } ctx.tx.peer = peer; sync_remote_cseq_with_request(ctx, &req); } @@ -2642,6 +2655,75 @@ mod tests { assert_eq!(second_req.header_value("CSeq"), Some("2 INVITE")); } + #[test] + fn session_refresh_keeps_update_when_response_omits_allow() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-update-retain", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: None, + }, + ); + + let (_peer, first_req) = parse_sent_request(&mut rx); + match first_req.method { + SipMethod::Update => {} + other => panic!("expected UPDATE, got {:?}", other), + } + + let resp = SipResponseBuilder::new(200, "OK") + .header("Call-ID", "call-update-retain") + .header( + "CSeq", + first_req + .header_value("CSeq") + .expect("outbound update CSeq"), + ) + .header("Contact", "") + .header("Session-Expires", "180;refresher=uas") + .build(); + let events = core.handle_response(resp, refreshed_peer()); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-update-retain" + && timer.expires == Duration::from_secs(180) + && timer.refresher == SessionRefresher::Uas + )); + let ctx = core.invites.get(&call_id).expect("invite context"); + assert!( + ctx.allow_update, + "missing Allow header must retain prior UPDATE support" + ); + assert_eq!(ctx.remote_target_uri, "sip:target-keep-update@example.net"); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(90), + local_sdp: None, + }, + ); + + let (peer, second_req) = parse_sent_request(&mut rx); + assert_eq!(peer, refreshed_peer()); + match second_req.method { + SipMethod::Update => {} + other => panic!("expected UPDATE after missing Allow, got {:?}", other), + } + assert_eq!(second_req.uri, "sip:target-keep-update@example.net"); + assert_eq!(second_req.header_value("CSeq"), Some("2 UPDATE")); + } + #[test] fn reinvite_refresh_missing_local_sdp_sends_bye_and_emits_failure() { let (mut core, mut rx) = test_core(); @@ -2797,6 +2879,163 @@ mod tests { assert_eq!(req.header_value("CSeq"), Some("1 UPDATE")); } + #[test] + fn inbound_update_without_allow_keeps_prior_update_support() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-update-no-allow", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + let update = SipRequestBuilder::new(SipMethod::Update, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-update-no-allow") + .header("CSeq", "77 UPDATE") + .header("Contact", "") + .header("Session-Expires", "180;refresher=uas") + .build(); + let events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: update.to_bytes(), + }); + + assert!(matches!( + events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-remote-update-no-allow" + && timer.expires == Duration::from_secs(180) + )); + let (_peer, resp) = parse_sent_response(&mut rx); + assert_eq!(resp.status_code, 200); + assert!( + core.invites + .get(&call_id) + .expect("invite context") + .allow_update, + "missing Allow on in-dialog UPDATE must retain prior UPDATE support" + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: None, + }, + ); + + let (_peer, req) = parse_sent_request(&mut rx); + match req.method { + SipMethod::Update => {} + other => panic!("expected UPDATE after missing Allow, got {:?}", other), + } + assert_eq!(req.header_value("CSeq"), Some("1 UPDATE")); + } + + #[test] + fn inbound_reinvite_without_allow_keeps_prior_update_support() { + let (mut core, _rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-reinvite-no-allow", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + let reinvite = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-reinvite-no-allow") + .header("CSeq", "99 INVITE") + .header("Contact", "") + .build(); + let events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: reinvite.to_bytes(), + }); + + assert!(matches!( + events.as_slice(), + [SipEvent::ReInvite { call_id, .. }] + if call_id.as_str() == "call-remote-reinvite-no-allow" + )); + let ctx = core.invites.get(&call_id).expect("invite context"); + assert!( + ctx.allow_update, + "missing Allow on in-dialog re-INVITE must retain prior UPDATE support" + ); + assert_eq!(ctx.remote_target_uri, "sip:alice-reinvite@example.com"); + } + + #[test] + fn delayed_request_does_not_roll_back_remote_cseq() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-cseq-order", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + let reinvite = SipRequestBuilder::new(SipMethod::Invite, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-cseq-order") + .header("CSeq", "99 INVITE") + .build(); + let reinvite_events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: reinvite.to_bytes(), + }); + assert!(matches!( + reinvite_events.as_slice(), + [SipEvent::ReInvite { call_id, .. }] if call_id.as_str() == "call-remote-cseq-order" + )); + assert_eq!( + core.invites + .get(&call_id) + .expect("invite context") + .remote_cseq, + Some(99) + ); + + let delayed_update = SipRequestBuilder::new(SipMethod::Update, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-cseq-order") + .header("CSeq", "77 UPDATE") + .header("Allow", "INVITE, ACK, UPDATE, BYE") + .header("Contact", "") + .header("Session-Expires", "180;refresher=uas") + .build(); + let update_events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: delayed_update.to_bytes(), + }); + assert!(matches!( + update_events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-remote-cseq-order" + && timer.expires == Duration::from_secs(180) + )); + let (_peer, resp) = parse_sent_response(&mut rx); + assert_eq!(resp.status_code, 200); + assert_eq!( + core.invites + .get(&call_id) + .expect("invite context") + .remote_cseq, + Some(99), + "delayed requests must not reduce the highest seen remote CSeq" + ); + } + #[test] fn reinvite_refresh_2xx_sends_ack_and_emits_session_refresh() { let (mut core, mut rx) = test_core(); From a635cb055c50087b2052f60e571c47c59f45a70e Mon Sep 17 00:00:00 2001 From: Masanori Suda Date: Wed, 11 Mar 2026 05:57:52 +0900 Subject: [PATCH 7/7] =?UTF-8?q?#280=20coderabiit=E6=8C=87=E6=91=98?= =?UTF-8?q?=E5=AF=BE=E5=BF=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/protocol/sip/core.rs | 268 ++++++++++++++++-- 1 file changed, 239 insertions(+), 29 deletions(-) diff --git a/virtual-voicebot-backend/src/protocol/sip/core.rs b/virtual-voicebot-backend/src/protocol/sip/core.rs index c7b1d6b..6a391e1 100644 --- a/virtual-voicebot-backend/src/protocol/sip/core.rs +++ b/virtual-voicebot-backend/src/protocol/sip/core.rs @@ -302,11 +302,7 @@ fn parse_min_se(value: &str) -> Option { } fn sync_remote_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { - if let Some(remote_cseq) = req - .header_value("CSeq") - .and_then(|value| parse_cseq_header(value).ok()) - .map(|cseq| cseq.num) - { + if let Some(remote_cseq) = request_cseq_number(req) { ctx.remote_cseq = Some( ctx.remote_cseq .map_or(remote_cseq, |existing| existing.max(remote_cseq)), @@ -314,6 +310,12 @@ fn sync_remote_cseq_with_request(ctx: &mut InviteContext, req: &SipRequest) { } } +fn request_cseq_number(req: &SipRequest) -> Option { + req.header_value("CSeq") + .and_then(|value| parse_cseq_header(value).ok()) + .map(|cseq| cseq.num) +} + fn request_remote_target_uri(req: &SipRequest) -> String { req.header_value("Contact") .map(extract_contact_uri) @@ -1263,10 +1265,7 @@ impl SipCore { ) -> Vec { // 再送判定 or re-INVITE 判定 if let Some(ctx) = self.invites.get_mut(&headers.call_id) { - let req_cseq = req - .header_value("CSeq") - .and_then(|value| parse_cseq_header(value).ok()) - .map(|cseq| cseq.num); + let req_cseq = request_cseq_number(&req); let prev_cseq = ctx.remote_cseq; if req_cseq.is_some() && prev_cseq.is_some() && req_cseq == prev_cseq { let (action, final_ok, tx_peer) = ( @@ -1447,7 +1446,9 @@ impl SipCore { if let Some(ctx) = self.invites.get_mut(&headers.call_id) { ctx.req = req.clone(); - ctx.remote_target_uri = request_remote_target_uri(&req); + if req.header_value("Contact").is_some() { + ctx.remote_target_uri = request_remote_target_uri(&req); + } if let Some(allow_update) = request_allows_update(&req) { ctx.allow_update = allow_update; } @@ -1791,26 +1792,35 @@ impl SipCore { self.send_payload(peer, bytes); } - if let Some(cfg) = session_timer { - if let Some(ctx) = self.invites.get_mut(&headers.call_id) { - ctx.session_timer = Some(cfg.clone()); - ctx.remote_target_uri = request_remote_target_uri(&req); + let mut refresh_event = None; + if let Some(ctx) = self.invites.get_mut(&headers.call_id) { + let req_cseq = request_cseq_number(&req); + let is_stale = matches!( + (req_cseq, ctx.remote_cseq), + (Some(incoming), Some(prev)) if incoming < prev + ); + if !is_stale { + if req.header_value("Contact").is_some() { + ctx.remote_target_uri = request_remote_target_uri(&req); + } if let Some(allow_update) = request_allows_update(&req) { ctx.allow_update = allow_update; } ctx.tx.peer = peer; sync_remote_cseq_with_request(ctx, &req); + if let Some(cfg) = session_timer.as_ref() { + ctx.session_timer = Some(cfg.clone()); + refresh_event = Some(SipEvent::SessionRefresh { + call_id: headers.call_id.clone(), + timer: SessionTimerInfo { + expires: cfg.expires, + refresher: cfg.refresher, + }, + }); + } } - vec![SipEvent::SessionRefresh { - call_id: headers.call_id, - timer: SessionTimerInfo { - expires: cfg.expires, - refresher: cfg.refresher, - }, - }] - } else { - vec![] } + refresh_event.into_iter().collect() } fn start_reliable_provisional( @@ -2935,6 +2945,153 @@ mod tests { assert_eq!(req.header_value("CSeq"), Some("1 UPDATE")); } + #[test] + fn inbound_update_without_session_expires_updates_dialog_state() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-update-no-session-expires", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + core.invites + .get_mut(&call_id) + .expect("invite context") + .remote_target_uri = "sip:target-before@example.net".to_string(); + + let update = SipRequestBuilder::new(SipMethod::Update, "sip:self-overwrite@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-update-no-session-expires") + .header("CSeq", "77 UPDATE") + .header("Allow", "INVITE, ACK, BYE") + .header("Contact", "") + .build(); + let events = core.handle_input(&SipInput { + peer: refreshed_peer(), + data: update.to_bytes(), + }); + + assert!( + events.is_empty(), + "UPDATE without Session-Expires is not a refresh" + ); + let (_peer, resp) = parse_sent_response(&mut rx); + assert_eq!(resp.status_code, 200); + let ctx = core.invites.get(&call_id).expect("invite context"); + assert_eq!(ctx.remote_cseq, Some(77)); + assert_eq!(ctx.tx.peer, refreshed_peer()); + assert_eq!(ctx.remote_target_uri, "sip:alice-refresh@example.com"); + assert!( + !ctx.allow_update, + "Allow without UPDATE should switch future refreshes to re-INVITE" + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: Some(Sdp::pcmu("127.0.0.1", 4000)), + }, + ); + + let (peer, req) = parse_sent_request(&mut rx); + assert_eq!(peer, refreshed_peer()); + match req.method { + SipMethod::Invite => {} + other => panic!( + "expected re-INVITE after UPDATE capability drop, got {:?}", + other + ), + } + assert_eq!(req.uri, "sip:alice-refresh@example.com"); + } + + #[test] + fn stale_update_with_session_expires_does_not_emit_refresh_or_roll_back_dialog_state() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-update-stale", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + + let current_update = SipRequestBuilder::new(SipMethod::Update, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-update-stale") + .header("CSeq", "99 UPDATE") + .header("Allow", "INVITE, ACK, UPDATE, BYE") + .header("Contact", "") + .header("Session-Expires", "180;refresher=uas") + .build(); + let current_events = core.handle_input(&SipInput { + peer: refreshed_peer(), + data: current_update.to_bytes(), + }); + assert!(matches!( + current_events.as_slice(), + [SipEvent::SessionRefresh { call_id, timer }] + if call_id.as_str() == "call-remote-update-stale" + && timer.expires == Duration::from_secs(180) + )); + let (_peer, resp) = parse_sent_response(&mut rx); + assert_eq!(resp.status_code, 200); + + let stale_update = SipRequestBuilder::new(SipMethod::Update, "sip:test@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-update-stale") + .header("CSeq", "77 UPDATE") + .header("Allow", "INVITE, ACK, BYE") + .header("Contact", "") + .header("Session-Expires", "90;refresher=uas") + .build(); + let stale_events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: stale_update.to_bytes(), + }); + assert!( + stale_events.is_empty(), + "stale UPDATE must not emit a new SessionRefresh" + ); + let (_peer, resp) = parse_sent_response(&mut rx); + assert_eq!(resp.status_code, 200); + + let ctx = core.invites.get(&call_id).expect("invite context"); + assert_eq!(ctx.remote_cseq, Some(99)); + assert_eq!(ctx.tx.peer, refreshed_peer()); + assert_eq!(ctx.remote_target_uri, "sip:alice-current@example.com"); + assert!(ctx.allow_update); + assert_eq!( + ctx.session_timer.as_ref().map(|cfg| cfg.expires), + Some(Duration::from_secs(180)) + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: None, + }, + ); + + let (peer, req) = parse_sent_request(&mut rx); + assert_eq!(peer, refreshed_peer()); + match req.method { + SipMethod::Update => {} + other => panic!( + "expected UPDATE after ignoring stale request, got {:?}", + other + ), + } + assert_eq!(req.uri, "sip:alice-current@example.com"); + } + #[test] fn inbound_reinvite_without_allow_keeps_prior_update_support() { let (mut core, _rx) = test_core(); @@ -2971,6 +3128,61 @@ mod tests { assert_eq!(ctx.remote_target_uri, "sip:alice-reinvite@example.com"); } + #[test] + fn inbound_reinvite_without_contact_keeps_prior_remote_target() { + let (mut core, mut rx) = test_core(); + let call_id = send_inbound_invite_with_cseq( + &mut core, + "call-remote-reinvite-no-contact", + Some("INVITE, ACK, UPDATE, BYE"), + 11, + ); + core.invites + .get_mut(&call_id) + .expect("invite context") + .remote_target_uri = "sip:target-before@example.net".to_string(); + + let reinvite = SipRequestBuilder::new(SipMethod::Invite, "sip:self-overwrite@example.com") + .header("Via", "SIP/2.0/UDP 127.0.0.1:5060") + .header("From", ";tag=alice") + .header("To", ";tag=rustbot") + .header("Call-ID", "call-remote-reinvite-no-contact") + .header("CSeq", "99 INVITE") + .build(); + let events = core.handle_input(&SipInput { + peer: dummy_peer(), + data: reinvite.to_bytes(), + }); + + assert!(matches!( + events.as_slice(), + [SipEvent::ReInvite { call_id, .. }] + if call_id.as_str() == "call-remote-reinvite-no-contact" + )); + assert_eq!( + core.invites + .get(&call_id) + .expect("invite context") + .remote_target_uri, + "sip:target-before@example.net" + ); + + core.handle_sip_command( + &call_id, + SipCommand::SendSessionRefresh { + expires: Duration::from_secs(120), + local_sdp: None, + }, + ); + + let (_peer, req) = parse_sent_request(&mut rx); + match req.method { + SipMethod::Update => {} + other => panic!("expected UPDATE, got {:?}", other), + } + assert_eq!(req.uri, "sip:target-before@example.net"); + } + #[test] fn delayed_request_does_not_roll_back_remote_cseq() { let (mut core, mut rx) = test_core(); @@ -3018,12 +3230,10 @@ mod tests { peer: dummy_peer(), data: delayed_update.to_bytes(), }); - assert!(matches!( - update_events.as_slice(), - [SipEvent::SessionRefresh { call_id, timer }] - if call_id.as_str() == "call-remote-cseq-order" - && timer.expires == Duration::from_secs(180) - )); + assert!( + update_events.is_empty(), + "stale UPDATE must not emit SessionRefresh" + ); let (_peer, resp) = parse_sent_response(&mut rx); assert_eq!(resp.status_code, 200); assert_eq!(