Use stored item types when materializing turn summaries (#39514)

## What changed

- Select user and agent summary items using the materialized `item_type` column.
- Fall back to the type in `item_json` when `item_type` is empty so rows written by older clients still produce the correct summary.

## Testing

- Extend the summary materialization test to cover items inserted without a stored item type.

GitOrigin-RevId: 45ca1107900e9b8d46561cea210a27ecd661cedc

codex-rs/thread-store/src/local/thread_history.rs

@@ -328,7 +328,10 @@ SET
             FROM thread_items
             WHERE thread_id = ?
               AND turn_id = ?
-              AND json_extract(item_json, '$.type') = 'userMessage'
+              AND (
+                item_type = 'userMessage'
+                OR (item_type = '' AND json_extract(item_json, '$.type') = 'userMessage')
+              )
             ORDER BY rollout_ordinal
             LIMIT 1
         )
@@ -339,7 +342,10 @@ SET
             FROM thread_items
             WHERE thread_id = ?
               AND turn_id = ?
-              AND json_extract(item_json, '$.type') = 'agentMessage'
+              AND (
+                item_type = 'agentMessage'
+                OR (item_type = '' AND json_extract(item_json, '$.type') = 'agentMessage')
+              )
               AND json_extract(item_json, '$.phase') = 'final_answer'
             ORDER BY rollout_ordinal DESC
             LIMIT 1
@@ -350,7 +356,10 @@ SET
                 FROM thread_items
                 WHERE thread_id = ?
                   AND turn_id = ?
-                  AND json_extract(item_json, '$.type') = 'agentMessage'
+                  AND (
+                    item_type = 'agentMessage'
+                    OR (item_type = '' AND json_extract(item_json, '$.type') = 'agentMessage')
+                  )
                   AND json_extract(item_json, '$.phase') IS NULL
                 ORDER BY rollout_ordinal DESC
                 LIMIT 1

codex-rs/thread-store/src/local/thread_history_materialization_tests.rs

@@ -1304,6 +1304,33 @@ async fn summary_items_use_final_answers_and_ignore_commentary() {
         })
         .await
         .expect("append items before turn lifecycle");
+
+    let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing(
+        home.path().abs(),
+    ))
+    .await
+    .expect("open thread history db");
+    sqlx::query(
+        r#"
+INSERT OR REPLACE INTO thread_items (
+    thread_id,
+    turn_id,
+    item_id,
+    rollout_ordinal,
+    created_at_ms,
+    item_json
+)
+SELECT thread_id, turn_id, item_id, rollout_ordinal, created_at_ms, item_json
+FROM thread_items
+WHERE thread_id = ? AND turn_id = ?
+        "#,
+    )
+    .bind(thread_id.to_string())
+    .bind("turn-1")
+    .execute(&pool)
+    .await
+    .expect("older writers can append items without a stored item type");
+
     store
         .append_items(AppendThreadItemsParams {
             thread_id,