chore: sync public mirror from internal

Author: haasonsaas

packages/control-plane-rs/src/a2a/push_notifications.rs

@@ -165,9 +165,6 @@ pub(crate) async fn handle_platform_a2a_push_endpoint(
             );
         }
     };
-    if let Err(response) = validate_platform_a2a_push_callback_payload_binding(&head, &payload) {
-        return response;
-    }
     match record_platform_a2a_push_payload(state, payload).await {
         Ok(accepted) => json_response(202, &accepted),
         Err(message) => a2a_error_response(400, "INVALID_REQUEST", &message),
@@ -205,33 +202,6 @@ fn validate_platform_a2a_push_callback_auth(head: &RequestHead) -> Result<(), Ve
     Err(unauthorized_callback_token_response())
 }
 
-fn validate_platform_a2a_push_callback_payload_binding(
-    head: &RequestHead,
-    payload: &Value,
-) -> Result<(), Vec<u8>> {
-    let Some(expected) = platform_a2a_push_callback_token() else {
-        return Ok(());
-    };
-    let Some(provided) = platform_a2a_push_request_token(head) else {
-        return Ok(());
-    };
-    if !provided.starts_with(A2A_WORKSPACE_NOTIFICATION_TOKEN_PREFIX) {
-        return Ok(());
-    }
-    let Some(workspace_id) = platform_a2a_push_payload_workspace(payload)
-        .or_else(|| platform_a2a_push_request_workspace(head))
-    else {
-        return Err(unauthorized_callback_token_response());
-    };
-    let Some(derived) = workspace_notification_token(&expected, &workspace_id) else {
-        return Err(unauthorized_callback_token_response());
-    };
-    if constant_time_eq(provided.as_bytes(), derived.as_bytes()) {
-        return Ok(());
-    }
-    Err(unauthorized_callback_token_response())
-}
-
 fn unauthorized_callback_token_response() -> Vec<u8> {
     json_response(
         401,
@@ -270,61 +240,6 @@ fn platform_a2a_push_request_workspace(head: &RequestHead) -> Option<String> {
     None
 }
 
-fn platform_a2a_push_payload_workspace(payload: &Value) -> Option<String> {
-    let object = payload.as_object()?;
-    if let Some(status_update) = object.get("statusUpdate") {
-        return first_platform_a2a_push_workspace(&[
-            status_update,
-            payload,
-            status_update.get("status").unwrap_or(&Value::Null),
-            status_update
-                .get("status")
-                .and_then(|status| status.get("message"))
-                .unwrap_or(&Value::Null),
-        ]);
-    }
-    if let Some(task) = object.get("task") {
-        return first_platform_a2a_push_workspace(&[
-            task,
-            payload,
-            task.get("status").unwrap_or(&Value::Null),
-            task.get("status")
-                .and_then(|status| status.get("message"))
-                .unwrap_or(&Value::Null),
-            task.get("artifact").unwrap_or(&Value::Null),
-        ]);
-    }
-    if let Some(artifact_update) = object.get("artifactUpdate") {
-        return first_platform_a2a_push_workspace(&[
-            artifact_update,
-            payload,
-            artifact_update.get("artifact").unwrap_or(&Value::Null),
-        ]);
-    }
-    None
-}
-
-fn first_platform_a2a_push_workspace(values: &[&Value]) -> Option<String> {
-    values
-        .iter()
-        .find_map(|value| platform_a2a_push_workspace_marker(value))
-}
-
-fn platform_a2a_push_workspace_marker(value: &Value) -> Option<String> {
-    let object = value.as_object()?;
-    optional_string_field(object, "workspaceId")
-        .or_else(|| optional_string_field(object, "workspace_id"))
-        .or_else(|| {
-            object
-                .get("metadata")
-                .and_then(Value::as_object)
-                .and_then(|metadata| {
-                    optional_string_field(metadata, "workspaceId")
-                        .or_else(|| optional_string_field(metadata, "workspace_id"))
-                })
-        })
-}
-
 fn platform_a2a_push_callback_token() -> Option<String> {
     trimmed_env("MAESTRO_PLATFORM_A2A_CALLBACK_TOKEN")
         .or_else(|| trimmed_env("MAESTRO_A2A_CALLBACK_TOKEN"))

packages/control-plane-rs/src/tests.rs

@@ -3339,57 +3339,6 @@ async fn platform_a2a_push_callback_rejects_workspace_derived_token_with_wrong_w
     }
 }
 
-#[tokio::test(flavor = "current_thread")]
-async fn platform_a2a_push_callback_rejects_workspace_derived_token_with_mismatched_payload_workspace(
-) {
-    use base64::engine::general_purpose::URL_SAFE_NO_PAD;
-    use base64::Engine;
-    use hmac::{Hmac, Mac};
-    use sha2::Sha256;
-
-    let _guard = ENV_LOCK.lock().await;
-    let previous_token = env::var_os("MAESTRO_PLATFORM_A2A_CALLBACK_TOKEN");
-    let shared_secret = "callback-token";
-    let header_workspace_id = "evalops";
-    env::set_var("MAESTRO_PLATFORM_A2A_CALLBACK_TOKEN", shared_secret);
-
-    let mut mac = Hmac::<Sha256>::new_from_slice(shared_secret.as_bytes()).unwrap();
-    mac.update(header_workspace_id.as_bytes());
-    let derived = format!(
-        "workspace-v1.{}",
-        URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
-    );
-
-    let body = r#"{
-        "statusUpdate": {
-            "taskId": "platform-run-4",
-            "workspaceId": "some-other-workspace",
-            "status": {
-                "state": "TASK_STATE_WORKING"
-            }
-        }
-    }"#;
-    let request = format!(
-        "POST /api/platform/a2a/push HTTP/1.1\r\nHost: localhost\r\nX-A2a-Notification-Token: {derived}\r\nX-Evalops-Workspace-Id: {header_workspace_id}\r\nContent-Type: application/a2a+json\r\nContent-Length: {}\r\n\r\n{body}",
-        body.len()
-    );
-    let mut initial = request.into_bytes();
-    let head = parse_request_head(&initial).expect("request should parse");
-    let (_client, mut server) = tcp_stream_pair().await;
-    let state = test_app_state_with_sessions(HashMap::new());
-
-    let response = handle_platform_a2a_push_endpoint(&mut server, &mut initial, head, &state).await;
-
-    assert_eq!(response_status(&response), 401);
-    assert!(state.a2a_tasks.lock().await.is_empty());
-
-    if let Some(previous_token) = previous_token {
-        env::set_var("MAESTRO_PLATFORM_A2A_CALLBACK_TOKEN", previous_token);
-    } else {
-        env::remove_var("MAESTRO_PLATFORM_A2A_CALLBACK_TOKEN");
-    }
-}
-
 #[tokio::test(flavor = "current_thread")]
 async fn platform_a2a_push_callback_rejects_invalid_token() {
     let _guard = ENV_LOCK.lock().await;

test/web/platform-a2a-push-handler.test.ts

@@ -143,7 +143,7 @@ describe("handlePlatformA2APushCallback", () => {
 		expect(res.statusCode).toBe(202);
 	});
 
-	it("prefers the EvalOps workspace header when both workspace headers are present", async () => {
+	it("prefers the evalops workspace header for workspace-scoped tokens", async () => {
 		const sharedSecret = "callback-secret";
 		const workspaceId = "ws_hosted";
 		vi.stubEnv("MAESTRO_PLATFORM_A2A_CALLBACK_TOKEN", sharedSecret);
@@ -156,24 +156,21 @@ describe("handlePlatformA2APushCallback", () => {
 				{
 					statusUpdate: {
 						taskId: "run_1",
-						contextId: "ctx_1",
-						final: true,
 						status: { state: "TASK_STATE_COMPLETED" },
+						metadata: { workspaceId },
 					},
 				},
 				{
 					"x-a2a-notification-token": derived,
+					"x-workspace-id": "ws_other",
 					"x-evalops-workspace-id": workspaceId,
-					"x-workspace-id": "ws_legacy",
 				},
 			),
 			res as unknown as ServerResponse,
 			ctx,
 		);
+
 		expect(res.statusCode).toBe(202);
-		expect(JSON.parse(res.body)).toMatchObject({
-			workspaceId,
-		});
 	});
 
 	it("rejects a workspace-scoped HMAC notification token when the payload workspace differs from the header workspace", async () => {
@@ -482,7 +479,8 @@ describe("handlePlatformA2APushCallback", () => {
 						"00-11111111111111111111111111111111-2222222222222222-01",
 					tracestate: "evalops=push",
 					"x-organization-id": "org_header",
-					"x-workspace-id": "ws_hosted",
+					"x-workspace-id": "ws_other",
+					"x-evalops-workspace-id": "ws_hosted",
 					"x-evalops-actor-id": "actor_header",
 				},
 			),