@@ -19,7 +19,7 @@ jobs:
     # A successful analysis job does not mean that no alerts were found.
     name: CodeQL (${{ matrix.language }})
     runs-on: ubuntu-latest
-    timeout-minutes: 45
+    timeout-minutes: ${{ matrix.language == 'rust' && 180 || 45 }}
     permissions:
       contents: read
       security-events: write
 
@@ -67,7 +67,11 @@
     {
       "path": "0017-native-policy-expiry-admission.patch",
       "sha256": "4406a07b8bcd9769595ba9295e10986311e8b3fd96bcfeda3d9dfe13a3095048"
+    },
+    {
+      "path": "0018-api-key-rate-limits.patch",
+      "sha256": "7214955c12f507f0534a7813205eeba656265aa6d919aa3704ec5a47f3ea7d64"
     }
   ],
-  "result_tree": "66f4dc7319790001e362f9dc425916708efe4729"
+  "result_tree": "031267b70bbf9fda8d45193acc5d41776c136c5a"
 }
@@ -6370,7 +6370,7 @@ pub(crate) mod tests {
             "name":"update_plan", "namespace":"functions",
             "arguments":"{\"plan\":[]}", "encrypted_function_args":[],
         }));
-        let original = serde_json::to_value(&item).unwrap();
+        assert_eq!(item.id().unwrap().as_str(), "fc_server");
         let upstream = Box::pin(stream::iter(vec![
             Ok(ResponseEvent::Created {
                 response_id: Some("real-provider".into()),
@@ -6386,7 +6386,7 @@ pub(crate) mod tests {
         let mut intercepted = Box::new(request).intercept(upstream);
         assert!(matches!(
             intercepted.next().await,
-            Some(Ok(ResponseEvent::Created { .. }))
+            Some(Ok(ResponseEvent::Created { response_id: Some(id) })) if id == "real-provider"
         ));
         let edit = async {
             let req = control.next().await.unwrap();
@@ -6404,16 +6404,21 @@ pub(crate) mod tests {
         let Some(Ok(ResponseEvent::OutputItemDone(item))) = event else {
             panic!("missing actual item")
         };
-        let mut actual = serde_json::to_value(item).unwrap();
+        assert!(item.id().is_none());
+        assert!(matches!(&item, ResponseItem::FunctionCall { call_id, .. }
+            if call_id == "real-native-call"));
         assert_eq!(
-            serde_json::from_str::<Value>(actual["arguments"].as_str().unwrap()).unwrap(),
-            json!({"plan":[],"explanation":"effective"})
+            serde_json::to_value(item).unwrap(),
+            json!({
+                "type": "function_call", "call_id": "real-native-call",
+                "name": "update_plan", "namespace": "functions",
+                "arguments": json!({"plan": [], "explanation": "effective"}).to_string(),
+                "encrypted_function_args": [],
+            })
         );
-        actual["arguments"] = original["arguments"].clone();
-        assert_eq!(actual, original); // including encrypted flags, id, name, namespace and metadata
         assert!(matches!(
             intercepted.next().await,
-            Some(Ok(ResponseEvent::Created { .. }))
+            Some(Ok(ResponseEvent::Created { response_id: Some(id) })) if id == "next-provider-event"
         ));
         assert!(intercepted.next().await.is_none());
     }
@@ -6477,7 +6482,7 @@ pub(crate) mod tests {
                 "arguments":r#"{"task_name":"old","message":"opaque"}"#,
                 "encrypted_function_args":metadata,
             }));
-            let original = serde_json::to_value(&item).unwrap();
+            assert_eq!(item.id().unwrap().as_str(), "fc_original");
             let request = direct_request(control.clone(), "turn");
             let edit = async {
                 let req = control.next().await.unwrap();
@@ -6488,13 +6493,20 @@ pub(crate) mod tests {
             };
             let (result, ()) = tokio::join!(request.before(item, None), edit);
             if accepted {
-                let mut actual = serde_json::to_value(result.unwrap().0).unwrap();
-                assert_eq!(
-                    serde_json::from_str::<Value>(actual["arguments"].as_str().unwrap()).unwrap(),
-                    value
-                );
-                actual["arguments"] = original["arguments"].clone();
-                assert_eq!(actual, original);
+                let (item, changed, _) = result.unwrap();
+                assert!(changed);
+                assert!(item.id().is_none());
+                assert!(matches!(&item, ResponseItem::FunctionCall { call_id, .. }
+                    if call_id == "original"));
+                let mut expected = json!({
+                    "type": "function_call", "call_id": "original",
+                    "name": "spawn_agent", "namespace": "collaboration",
+                    "arguments": value.to_string(),
+                });
+                if !metadata.is_null() {
+                    expected["encrypted_function_args"] = metadata;
+                }
+                assert_eq!(serde_json::to_value(item).unwrap(), expected);
             } else {
                 assert!(
                     matches!(result, Err(ModelResponseError::InvalidRequest { message })
 
@@ -422,6 +422,7 @@ params!(TurnParams {
     #[serde(default)] mode: Mode,
     expected_turn_id: Option<String>,
     expected_previous_turn_id: Option<String>,
+    #[serde(default)] exact_steering: bool,
 });
 
 fn rejection(reason: NotSubmittedReason) -> Value {
@@ -561,6 +562,10 @@ async fn start_with_params<'a>(
         &mut Option<tokio::sync::MutexGuard<'a, ()>>,
     )>,
 ) -> Result<Value> {
+    ensure!(
+        !p.exact_steering || (managed.is_some() && matches!(p.mode, Mode::Steer)),
+        "exact steering requires a managed steer"
+    );
     let input = if p.input.is_array() {
         TurnInput::UserInput {
             content: decode::<Vec<UserInput>>(p.input)?,
@@ -659,6 +664,7 @@ async fn start_with_params<'a>(
             let expected = p
                 .expected_turn_id
                 .context("steer requires expected_turn_id")?;
+            let mut input_id = None;
             if managed.is_some()
                 && let TurnInput::UserInput { client_id, .. } = &mut request.input
             {
@@ -667,12 +673,19 @@ async fn start_with_params<'a>(
                     .get::<crate::python_tools::PythonTools>()
                     .context("Managed steering lost its tool owner")?
                     .admission_ready(&expected)?;
-                *client_id = Some(receipt.steering_client_id());
+                let id = receipt.root_steering_client_id(p.exact_steering)?;
+                if p.exact_steering {
+                    input_id = Some(id.clone());
+                }
+                *client_id = Some(id);
             }
             Ok(match thread.steer_turn(request, expected).await? {
-                SteerSubmission::Steered { turn_id } => {
-                    json!({"status": "steered", "turn_id": turn_id})
-                }
+                SteerSubmission::Steered { turn_id } => match input_id {
+                    Some(input_id) => {
+                        json!({"status": "steered", "turn_id": turn_id, "input_id": input_id})
+                    }
+                    None => json!({"status": "steered", "turn_id": turn_id}),
+                },
                 SteerSubmission::NotSubmitted { reason } => rejection(reason),
             })
         }
 
@@ -216,6 +216,7 @@ pub(crate) struct TurnAdmission {
     // Private namespace for SDK steers of this exact runtime, including same-ID recovery.
     steering_namespace: String,
     steering_tail: AtomicBool,
+    exact_steering: OnceLock<bool>,
     provider_error: Mutex<Option<(Value, Value)>>,
     policy_failure: OnceLock<&'static str>,
 }
@@ -304,6 +305,18 @@ impl TurnAdmission {
         )
     }
 
+    pub(crate) fn root_steering_client_id(&self, exact: bool) -> Result<String> {
+        ensure!(
+            *self.exact_steering.get_or_init(|| exact) == exact,
+            "Cannot mix exact and continuation steering on a native turn"
+        );
+        Ok(if exact {
+            format!("agents-exact-steer:{}", codex_protocol::ThreadId::new())
+        } else {
+            self.steering_client_id()
+        })
+    }
+
     fn record_steering(&self, client_id: Option<&str>) {
         if client_id.is_some_and(|id| id.starts_with(&self.steering_namespace)) {
             self.steering_tail.store(true, Ordering::Release);
@@ -2190,6 +2203,7 @@ impl TurnLifecycleContributor for PythonContributor {
                         codex_protocol::ThreadId::new()
                     ),
                     steering_tail: AtomicBool::new(false),
+                    exact_steering: OnceLock::new(),
                     provider_error: Mutex::default(),
                     policy_failure: OnceLock::new(),
                 });
 
@@ -56,6 +56,10 @@
         ReplaceArguments as ReplaceArguments,
         StopTurn as StopTurn,
     )
+    from .rate_limits import (
+        ApiRateLimits as ApiRateLimits,
+        ApiRateLimitWindow as ApiRateLimitWindow,
+    )
     from .schema import (
         NativeOutput as NativeOutput,
         OutputSpec as OutputSpec,
@@ -95,7 +99,12 @@
         ToolRef as ToolRef,
         WebSearch as WebSearch,
     )
-    from .turn import PendingRequests as PendingRequests, Turn as Turn
+    from .turn import (
+        PendingRequests as PendingRequests,
+        SteeringReceipt as SteeringReceipt,
+        Turn as Turn,
+        TurnNeedsAttention as TurnNeedsAttention,
+    )
 
 _MODULES = {
     "Allow": ".native_interception",
@@ -133,6 +142,8 @@
     "OutputValidationError": ".schema",
     "CheckpointStatus": ".state",
     "LocalStorage": ".state",
+    "ApiRateLimits": ".rate_limits",
+    "ApiRateLimitWindow": ".rate_limits",
     "InputImage": ".input",
     "InputText": ".input",
     "UserMessage": ".input",
@@ -166,6 +177,8 @@
     "SessionBusyError": ".session",
     "Turn": ".turn",
     "PendingRequests": ".turn",
+    "SteeringReceipt": ".turn",
+    "TurnNeedsAttention": ".turn",
     "Subagent": ".subagents",
 }
 
 
@@ -13,12 +13,14 @@
 from dataclasses import dataclass, field, fields, replace
 from typing import TYPE_CHECKING, Any, Literal, TypeAlias, TypeVar, cast
 
+from .rate_limits import ApiRateLimits
 from .usage import TokenUsageInfo
 
 if TYPE_CHECKING:
     from .turn import Turn
 
 __all__ = [
+    "ApiRateLimitsUpdated",
     "ContextCompacted",
     "EventGap",
     "EventStream",
@@ -86,6 +88,14 @@ class UsageUpdated(_SubmissionEvent):
     type: Literal["usage_updated"] = field(default="usage_updated", init=False)
 
 
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ApiRateLimitsUpdated(_SubmissionEvent):
+    """Rate-limit fields reported on a provider response for this turn."""
+
+    snapshot: ApiRateLimits
+    type: Literal["api_rate_limits_updated"] = field(default="api_rate_limits_updated", init=False)
+
+
 @dataclass(frozen=True, slots=True, kw_only=True)
 class ContextCompacted(_SubmissionEvent):
     """Core completed this native context-compaction item."""
@@ -164,6 +174,7 @@ class EventGap(_Event):
     | SubagentLifecycle
     | OutputTextDelta
     | UsageUpdated
+    | ApiRateLimitsUpdated
     | ContextCompacted
     | ToolStarted
     | ToolFinished
 
@@ -0,0 +1,116 @@
+"""Rate limits reported by the API-key model provider."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from decimal import Context, Decimal, localcontext
+from email.utils import parsedate_to_datetime
+from typing import Any
+
+__all__ = ["ApiRateLimitWindow", "ApiRateLimits"]
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ApiRateLimitWindow:
+    """Reported counts and time until reset, measured at the provider response.
+
+    A missing field means the response did not supply a usable value.
+    """
+
+    limit: int | None = None
+    remaining: int | None = None
+    reset_after_seconds: Decimal | None = None
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ApiRateLimits:
+    """Rate limits from the latest observed response with usable rate-limit data.
+
+    Retry advice can be a delay in seconds or an absolute UTC time. The
+    provider reports it separately from the remaining request and token counts.
+    """
+
+    requests: ApiRateLimitWindow | None = None
+    tokens: ApiRateLimitWindow | None = None
+    project_tokens: ApiRateLimitWindow | None = None
+    retry_after_seconds: int | None = None
+    retry_at: datetime | None = None
+
+
+_DURATION_UNIT_SECONDS = {
+    "h": Decimal(3600),
+    "m": Decimal(60),
+    "s": Decimal(1),
+    "ms": Decimal("0.001"),
+    "us": Decimal("0.000001"),
+    "ns": Decimal("0.000000001"),
+}
+_DURATION_COMPONENT = re.compile(r"([0-9]+(?:\.[0-9]+)?)(ms|us|ns|h|m|s)")
+_DURATION = re.compile(r"(?:[0-9]+(?:\.[0-9]+)?(?:ms|us|ns|h|m|s))+")
+
+
+def _duration_seconds(value: Any) -> Decimal | None:
+    if not isinstance(value, str) or len(value) > 64 or not _DURATION.fullmatch(value):
+        return None
+    # The input length bounds the needed precision. Keep the application's
+    # decimal context from rounding a provider value.
+    with localcontext(Context(prec=80)):
+        return sum(
+            (
+                Decimal(amount) * _DURATION_UNIT_SECONDS[unit]
+                for amount, unit in _DURATION_COMPONENT.findall(value)
+            ),
+            start=Decimal(0),
+        )
+
+
+def _count(value: Any) -> int | None:
+    if type(value) is int and value >= 0:
+        return value
+    return None
+
+
+def _window(value: Any) -> ApiRateLimitWindow | None:
+    if not isinstance(value, dict):
+        return None
+    limit = _count(value.get("limit"))
+    remaining = _count(value.get("remaining"))
+    reset = _duration_seconds(value.get("reset_after"))
+    if limit is None and remaining is None and reset is None:
+        return None
+    return ApiRateLimitWindow(limit=limit, remaining=remaining, reset_after_seconds=reset)
+
+
+def _retry_after(value: Any) -> tuple[int | None, datetime | None]:
+    if not isinstance(value, str) or len(value) > 64:
+        return None, None
+    if value.isascii() and value.isdecimal():
+        return int(value), None
+    try:
+        at = parsedate_to_datetime(value)
+        if at.tzinfo is None:
+            at = at.replace(tzinfo=timezone.utc)
+        return None, at.astimezone(timezone.utc)
+    except (ValueError, OverflowError):
+        return None, None
+
+
+def _api_rate_limits(value: Any) -> ApiRateLimits | None:
+    """Project the native wire snapshot without copying unrecognized fields."""
+    if not isinstance(value, dict):
+        return None
+    requests = _window(value.get("requests"))
+    tokens = _window(value.get("tokens"))
+    project_tokens = _window(value.get("project_tokens"))
+    delay, retry_at = _retry_after(value.get("retry_after"))
+    if all(field is None for field in (requests, tokens, project_tokens, delay, retry_at)):
+        return None
+    return ApiRateLimits(
+        requests=requests,
+        tokens=tokens,
+        project_tokens=project_tokens,
+        retry_after_seconds=delay,
+        retry_at=retry_at,
+    )
Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,11 @@`
`67`	`67`	`{`
`68`	`68`	`"path": "0017-native-policy-expiry-admission.patch",`
`69`	`69`	`"sha256": "4406a07b8bcd9769595ba9295e10986311e8b3fd96bcfeda3d9dfe13a3095048"`
	`70`	`+ },`
	`71`	`+ {`
	`72`	`+ "path": "0018-api-key-rate-limits.patch",`
	`73`	`+ "sha256": "7214955c12f507f0534a7813205eeba656265aa6d919aa3704ec5a47f3ea7d64"`
`70`	`74`	`}`
`71`	`75`	`],`
`72`		`- "result_tree": "66f4dc7319790001e362f9dc425916708efe4729"`
	`76`	`+ "result_tree": "031267b70bbf9fda8d45193acc5d41776c136c5a"`
`73`	`77`	`}`