Handle ChatGPT subscription metadata without aborting streams

This commit is contained in:
Ryan Ward
2026-10-19 19:41:37 +00:00
parent c894a11589
commit 75b9334a0b
3 changed files with 82 additions and 4 deletions
+4 -2
View File
@@ -5,6 +5,7 @@ This file provides guidance when working with code in this repository.
## Development Commands ## Development Commands
### Build and Run ### Build and Run
- `cargo build` - Build Galaxy and verify feature and bug-fix changes
- `cargo run` - Build and run Warp locally - `cargo run` - Build and run Warp locally
- `cargo bundle --bin warp` - Bundle the main app - `cargo bundle --bin warp` - Bundle the main app
- `make run` - Build and run `galaxy-oss` - `make run` - Build and run `galaxy-oss`
@@ -28,8 +29,9 @@ Environment variables:
- `WS_SERVER_URL` - WebSocket endpoint (default: `ws://localhost:8080/graphql/v2`) - `WS_SERVER_URL` - WebSocket endpoint (default: `ws://localhost:8080/graphql/v2`)
### Testing ### Testing
- Before launching `galaxy-oss`, check for existing app instances and close the prior development instance so only one app instance runs at a time. Distinguish the app process from its `terminal-server` helper; do not count the helper as a duplicate app. Leave the newly launched instance running for the user. - This development machine is CLI-only and has no GUI. The user performs GUI and most behavioral testing on a remote machine.
- During interactive feature and bug-fix verification, make the requested code changes first and run only `cargo run --bin galaxy-oss` for the user to verify. Leave the app running for the user; do not stop it or treat the command timeout as a failure. Do not run `cargo fmt`, `cargo check`, `cargo test`, `cargo clippy`, presubmit, or other validation commands until the user confirms the behavior. Run formatting, tests, and linting only as the final cleanup step after interactive approval. - Verify feature and bug-fix changes with `cargo build` only. Do not use `cargo run`, `make run`, or launch the app to verify results, and do not wait for interactive approval before completing the work.
- After `cargo build` succeeds and all requested work and final review are complete, commit the changes and push the current branch to both configured remotes, `origin` and `samsung`, using ordinary non-force pushes. If the build fails, report the blocker and do not push.
- `cargo nextest run --no-fail-fast --workspace --exclude command-signatures-v2` - Run tests with nextest - `cargo nextest run --no-fail-fast --workspace --exclude command-signatures-v2` - Run tests with nextest
- `cargo nextest run -p galaxy_completer --features v2` - Run completer tests with v2 features - `cargo nextest run -p galaxy_completer --features v2` - Run completer tests with v2 features
- `cargo test --doc` - Run doc tests - `cargo test --doc` - Run doc tests
+71 -2
View File
@@ -1,13 +1,14 @@
use galaxy_agent_core::StopReason;
use rig_core::client::CompletionClient; use rig_core::client::CompletionClient;
use rig_core::completion::{AssistantContent, CompletionModel, Message}; use rig_core::completion::{AssistantContent, CompletionModel, Message};
use rig_core::message::{ToolResultContent, UserContent}; use rig_core::message::{ToolResultContent, UserContent};
use rig_core::providers::chatgpt::ChatGPTAuth; use rig_core::providers::chatgpt::ChatGPTAuth;
use rig_core::test_utils::RecordingHttpClient; use rig_core::test_utils::{RecordingHttpClient, SequencedStreamingHttpClient};
use super::*; use super::*;
const COMPLETED_RESPONSE: &str = r#"data: {"type":"response.output_text.delta","delta":"ok"} const COMPLETED_RESPONSE: &str = r#"data: {"type":"response.output_text.delta","delta":"ok"}
data: {"type":"response.completed","response":{"id":"resp_test","object":"response","created_at":1,"status":"completed","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"gpt-5.3-codex","usage":{"input_tokens":1,"input_tokens_details":{"cached_tokens":0},"output_tokens":1,"output_tokens_details":{"reasoning_tokens":0},"total_tokens":2},"output":[{"type":"message","id":"msg_test","status":"completed","role":"assistant","content":[{"type":"output_text","annotations":[],"text":"ok"}]}],"tools":[]}} data: {"type":"response.completed","sequence_number":54,"response":{"id":"resp_test","object":"response","created_at":1,"status":"completed","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"gpt-5.3-codex","usage":{"input_tokens":1,"input_tokens_details":{"cached_tokens":0},"output_tokens":1,"output_tokens_details":{"reasoning_tokens":0},"total_tokens":2},"output":[{"type":"message","id":"msg_test","status":"completed","role":"assistant","content":[{"type":"output_text","annotations":[],"text":"ok"}]}],"tools":[]}}
data: [DONE]"#; data: [DONE]"#;
#[test] #[test]
@@ -151,3 +152,71 @@ async fn chatgpt_follow_up_request_preserves_responses_call_ids() {
assert_eq!(input[0]["call_id"], "call_native_1"); assert_eq!(input[0]["call_id"], "call_native_1");
assert_eq!(input[1]["call_id"], "call_native_1"); assert_eq!(input[1]["call_id"], "call_native_1");
} }
#[tokio::test]
async fn chatgpt_metadata_does_not_interrupt_text_or_completion() {
let metadata = r#"data: {"type":"response.metadata","response_id":"resp_test","sequence_number":51,"metadata":{"openai_chatgpt_moderation_metadata":{"prompt":{},"generation":{},"tool_call":{},"tool_response":{}}}}"#;
let delta = serde_json::json!({
"type": "response.output_text.delta",
"item_id": "msg_test",
"output_index": 0,
"content_index": 0,
"sequence_number": 52,
"delta": "ok",
});
let completed = COMPLETED_RESPONSE
.lines()
.find(|line| line.starts_with("data: {\"type\":\"response.completed\""))
.expect("fixture should contain a completed response");
let response =
format!("{metadata}\n\ndata: {delta}\n\n{metadata}\n\n{completed}\n\ndata: [DONE]\n\n");
let client = chatgpt::Client::builder()
.api_key(ChatGPTAuth::AccessToken {
access_token: "test-token".to_string(),
account_id: None,
})
.http_client(SequencedStreamingHttpClient::new(vec![Ok(response.into())]))
.build()
.expect("client should build");
let request = build_completion_request(
TurnRequest::new(
"gpt-5.3-codex".to_string(),
vec![ConversationMessage {
role: MessageRole::User,
content: MessageContent::Text("hello".to_string()),
}],
),
None,
true,
true,
None,
)
.expect("request should convert");
let (_sender, control) = turn_control();
let mut stream = start_model_turn(
client.completion_model("gpt-5.3-codex"),
request,
control,
None,
)
.await
.expect("request should reach the mocked provider");
let mut text = String::new();
let mut usage = None;
let mut stop_reason = None;
while let Some(event) = stream.next().await {
match event.expect("metadata should not fail the turn") {
AgentEvent::TextDelta { text: delta } => text.push_str(&delta),
AgentEvent::UsageUpdated { usage: updated } => usage = Some(updated),
AgentEvent::TurnStopped { reason } => stop_reason = Some(reason),
_ => {}
}
}
assert_eq!(text, "ok");
assert_eq!(
usage.expect("completion should report usage").output_tokens,
1
);
assert_eq!(stop_reason, Some(StopReason::Completed));
}
+7
View File
@@ -160,6 +160,13 @@ where
}); });
} }
Ok(StreamedAssistantContent::Unknown(value)) => { Ok(StreamedAssistantContent::Unknown(value)) => {
// ChatGPT subscription streams include informational moderation
// metadata that Rig does not map to assistant content.
if value.get("type").and_then(serde_json::Value::as_str)
== Some("response.metadata")
{
continue;
}
if is_keepalive_event(&value) { if is_keepalive_event(&value) {
yield Ok(AgentEvent::KeepAlive); yield Ok(AgentEvent::KeepAlive);
} else { } else {